node:http: an Upgrade request with a body follows Node v26

packages/bun-uws/src/HttpContext.h

@@ -37,7 +37,7 @@
 
 
 extern "C" void Bun__NodeHTTP__onReadsResumable(int ssl, struct us_socket_t *s);
-extern "C" void Bun__NodeHTTP__onReadParsed(int ssl, struct us_socket_t *s);
+extern "C" void Bun__NodeHTTP__onReadParsed(int ssl, struct us_socket_t *s, const char *head, unsigned int length, bool complete);
 
 namespace uWS {
 
@@ -336,7 +336,7 @@ struct HttpContext {
 
         /* Do not accept any data while in shutdown state. A node:http tunnel is half-open: it still reads after socket.end(). */
         bool isHalfOpenTunnel = false;
-        if constexpr (IsNodeHttp) isHalfOpenTunnel = httpResponseData->isConnectRequest;
+        if constexpr (IsNodeHttp) isHalfOpenTunnel = httpResponseData->isConnectRequest || (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY);
         if (us_socket_is_shut_down((us_socket_t *) s) && !isHalfOpenTunnel) {
             /* Balance the us_socket_ref above — every other return path
              * reaches the unref via returnedData. */
@@ -598,8 +598,8 @@ struct HttpContext {
                 return nullptr;
             }
 
-            /* We absolutely have to terminate parsing if shutdown */
-            if (us_socket_is_shut_down((us_socket_t *) s)) {
+            /* We absolutely have to terminate parsing if shutdown. A node:http Upgrade request reads on: its body, then its tunnel. */
+            if (us_socket_is_shut_down((us_socket_t *) s) && !(IsNodeHttp && (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY))) {
                 return nullptr;
             }
 
@@ -690,8 +690,10 @@ struct HttpContext {
                     return nullptr;
                 }
 
-                /* We absolutely have to terminate parsing if shutdown */
-                if (us_socket_is_shut_down((us_socket_t *) user)) {
+                /* We absolutely have to terminate parsing if shutdown. A node:http Upgrade request reads on: its body, then its tunnel. */
+                bool upgradeReadsOn = false;
+                if constexpr (IsNodeHttp) upgradeReadsOn = switchToTunnelAfterThisChunk || (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY);
+                if (us_socket_is_shut_down((us_socket_t *) user) && !upgradeReadsOn) {
                     /* node:http: this socket can be a tunnel that still reads. Its bytes are not for a body that ended. */
                     if constexpr (IsNodeHttp) {
                         if (fin) {
@@ -714,6 +716,10 @@ struct HttpContext {
                     if (us_socket_is_closed((struct us_socket_t *) user)) {
                         return nullptr;
                     }
+                    /* An 'upgrade' event waits for the end of this read: what follows the body is its head, not data of the tunnel. */
+                    if (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_NOTIFY_READ_PARSED) {
+                        return nullptr;
+                    }
                 }
             }
             /* JavaScript destroyed this socket earlier in this read: its message is delivered, so close before the next request is parsed. */
@@ -727,6 +733,25 @@ struct HttpContext {
             return user;
         });
 
+        /* node:http: the parked 'upgrade' (notifyWhenReadParsed) runs here, still in the parse window like a request handler. */
+        if constexpr (IsNodeHttp) {
+            if (httpContextData->upgradedWebSocket != s && !us_socket_is_closed(s)
+                && (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_NOTIFY_READ_PARSED)) [[unlikely]] {
+                httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_NODE_NOTIFY_READ_PARSED;
+                /* The data handler stopped the parse behind a body that ended in this read: the rest is the head. */
+                const bool complete = httpResponseData->isConnectRequest;
+                const unsigned int headStart = complete ? std::min(result.consumedBytes(), (unsigned int) length) : (unsigned int) length;
+                Bun__NodeHTTP__onReadParsed(SSL, s, data + headStart, (unsigned int) length - headStart, complete);
+                if (httpContextData->upgradedWebSocket || us_socket_is_closed(s)) {
+                    /* No HttpResponseData is left. A new WebSocket gets the head below. */
+                    result = HttpParserResult::success(headStart);
+                } else if (complete) {
+                    /* Balance the ref of onData, like a read that is parsed to its end. */
+                    us_socket_unref(s);
+                }
+            }
+        }
+
         auto httpErrorStatusCode = result.httpErrorStatusCode();
 
         /* Mark that we are no longer parsing Http */
@@ -791,18 +816,6 @@ struct HttpContext {
             /* We don't want open sockets to keep the event loop alive between HTTP requests */
             us_socket_unref((us_socket_t *) returnedData);
 
-            if constexpr (IsNodeHttp) {
-                if (httpResponseData->state & HttpResponseData<SSL>::HTTP_NODE_NOTIFY_READ_PARSED) {
-                    httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_NODE_NOTIFY_READ_PARSED;
-                    Bun__NodeHTTP__onReadParsed(SSL, (us_socket_t *) returnedData);
-                    /* That ran JavaScript: a closed or upgraded socket no longer has an HttpResponseData. */
-                    if (us_socket_is_closed((us_socket_t *) returnedData) || us_socket_kind((us_socket_t *) returnedData) != socketKind()) {
-                        ((AsyncSocket<SSL> *) returnedData)->uncork();
-                        return (us_socket_t *) returnedData;
-                    }
-                }
-            }
-
             /* node:http compat: a partial request head was left in the fallback
              * buffer by this read (either fresh bytes on an idle connection or a
              * pipelined request after the previous message completed) - its

src/js/internal/http.ts

@@ -30,7 +30,6 @@ const eofInProgress = Symbol("eofInProgress");
 const fakeSocketSymbol = Symbol("fakeSocket");
 const isTlsSymbol = Symbol("is_tls");
 const kHandle = Symbol("handle");
-const kOnReadParsed = Symbol("kOnReadParsed");
 const kRealListen = Symbol("kRealListen");
 const noBodySymbol = Symbol("noBody");
 const optionsSymbol = Symbol("options");
@@ -73,7 +72,6 @@ export const enum NodeHTTPResponseAbortEvent {
   none = 0,
   abort = 1,
   timeout = 2,
-  readParsed = 3,
 }
 export const enum NodeHTTPBodyReadState {
   none,
@@ -173,15 +171,6 @@ function emitEOFIncomingMessage(self) {
 }
 
 function onDataIncomingMessage(this: any, chunk, isLast, aborted: NodeHTTPResponseAbortEvent) {
-  if (aborted === NodeHTTPResponseAbortEvent.readParsed) {
-    const socket = this.socket;
-    const onReadParsed = socket?.[kOnReadParsed];
-    if (onReadParsed) {
-      socket[kOnReadParsed] = undefined;
-      onReadParsed(socket);
-    }
-    return;
-  }
   if (aborted === NodeHTTPResponseAbortEvent.abort) {
     // The request is aborted from the socket's #onClose (like Node.js's
     // socketOnClose → abortIncoming), which the native close path always
@@ -540,6 +529,7 @@ export {
   drainMicrotasks,
   emitCloseNT,
   emitEOFIncomingMessage,
+  emitEOFIncomingMessageOuter,
   emitErrorNextTickIfErrorListenerNT,
   eofInProgress,
   fakeSocketSymbol,
@@ -557,7 +547,6 @@ export {
   kHeaderSource,
   kInternalSocketData,
   kNeedDrain,
-  kOnReadParsed,
   kOutHeaders,
   kPendingCallbacks,
   kPerRequestCheckServerIdentity,

src/js/node/_http_server.ts

@@ -35,7 +35,6 @@ const {
   kInternalSocketData,
   serverSymbol,
   kHandle,
-  kOnReadParsed,
   kHandoffResponse,
   kHasQueuedReads,
   kRealListen,
@@ -51,6 +50,8 @@ const {
   NodeHTTPResponseFlags,
   callCloseCallback,
   emitCloseNT,
+  emitEOFIncomingMessageOuter,
+  eofInProgress,
   NodeHTTPResponseAbortEvent,
   STATUS_CODES,
   isTlsSymbol,
@@ -1047,7 +1048,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           // 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
+          // message completes; until then the upgradeHead is empty and everything after
           // the end of the message reaches the socket as raw data.
           socketHandle.upgradeToTunnel(hasBody, handle);
           socket[kHandoffResponse] = handle;
@@ -1056,12 +1057,14 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           // Node frees the parser before emitting 'upgrade' (socket.parser === null there).
           releaseServerParserShim(socket, http_req);
           if (hasBody) {
+            if ((dispatchBits & DISPATCH_READ_HAS_MORE) !== 0)
+              return parkUpgrade(server, socket, socketHandle, http_req, handle);
             socket[kUpgradeIncoming] = http_req;
             http_req.once("end", clearUpgradeIncoming.bind(undefined, socket));
           } else {
             http_req.complete = true;
           }
-          const upgradeHead = !hasBody && connectHead ? connectHead : kEmptyBuffer;
+          const upgradeHead = connectHead ? connectHead : kEmptyBuffer;
           socket.bytesRead += upgradeHead.length;
           let upgradeHandled;
           try {
@@ -1466,18 +1469,6 @@ function clearUpgradeIncoming(socket) {
   socket[kUpgradeIncoming] = undefined;
 }
 
-// Node's UpgradeStream._read: resume the request only while its body is incomplete.
-function resumePausedUpgradeIncoming(socket) {
-  const req = socket[kUpgradeIncoming];
-  if (req === undefined) return;
-  const response = socket[kHandle]?.response;
-  if (response && (response.hasBody & NodeHTTPBodyReadState.done) !== 0) {
-    socket[kUpgradeIncoming] = undefined;
-  } else {
-    req.resume();
-  }
-}
-
 // Node.js hands the connection over to 'connect'/'upgrade' listeners with the
 // connection-listener set removed (onParserExecuteCommon removes its data/end/
 // close/drain/error/timeout listeners) and only net.Socket's own 'end' listener
@@ -1493,6 +1484,41 @@ function detachSocketListenersForHandoff(socket) {
 function resolveHandoffPromise(promise) {
   $resolvePromise(promise, undefined);
 }
+
+// Node emits 'upgrade' once its parser has consumed the read, not at the end of the headers: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L1098-L1160
+function parkUpgrade(server, socket, socketHandle, req, handle) {
+  const promise = $newPromise();
+  handle.ondata = onDataParkedUpgrade.bind(req);
+  const handOff = handOffParkedUpgrade.bind(undefined, server, socket, req, handle, promise);
+  // Not armed: the socket closed in this dispatch, and no read of it is parsed on.
+  if (!socketHandle.notifyWhenReadParsed(handOff)) process.nextTick(handOff);
+  return promise;
+}
+
+// No EOF tick for the last chunk, or for a read() inside its 'data' (it would end a flowing request before the listener): handOffParkedUpgrade() completes the request.
+function onDataParkedUpgrade(this: import("node:http").IncomingMessage, chunk, isLast, aborted) {
+  if (isLast && aborted === NodeHTTPResponseAbortEvent.none) this[eofInProgress] = true;
+  onDataIncomingMessage.$call(this, chunk, false, aborted);
+}
+
+// head: what follows the body in the read. complete: the body ended in it.
+function handOffParkedUpgrade(server, socket, req, handle, promise, head: Buffer = kEmptyBuffer, complete = false) {
+  if (complete || req[eofInProgress]) {
+    emitEOFIncomingMessageOuter(req);
+  } else {
+    // The body is still arriving: a read of the socket also drives the request, like Node's UpgradeStream.
+    handle.ondata = onDataIncomingMessage.bind(req);
+    socket[kUpgradeIncoming] = req;
+    req.once("end", clearUpgradeIncoming.bind(undefined, socket));
+  }
+  socket.bytesRead += head.length;
+  try {
+    // No dispatch is on the stack: a listener that throws is an uncaught exception at once, as in Node.js.
+    if (!server.emit("upgrade", req, socket, head)) socket.destroy();
+  } finally {
+    socket.once("close", resolveHandoffPromise.bind(undefined, promise));
+  }
+}
 const kSocketTimeoutTimer = Symbol("socketTimeoutTimer");
 const kStreamingEnabled = Symbol("kStreamingEnabled");
 // destroySoon() was called: native lets the read that is being parsed reach the request, then closes right behind the FIN.
@@ -1667,7 +1693,6 @@ function getNodeHTTPServerSocket() {
     [kClosedWithUnsentData] = undefined;
     [kHandle];
     [kUpgradeIncoming]: import("node:http").IncomingMessage | undefined = undefined;
-    [kOnReadParsed] = undefined;
     [kHandoffResponse] = undefined;
     [kDestroySoon] = false;
     [kHandedOff] = false;
@@ -2069,9 +2094,9 @@ function getNodeHTTPServerSocket() {
         response?.resume();
         // Not paused: req.complete is false in the listener, so it can wait for 'end' with no reader.
         if (upgradeIncoming.readableFlowing !== false) upgradeIncoming.resume();
-        // The 'upgrade' listener runs before the rest of its read is parsed; that rest can complete the body.
-        else if (response?.notifyWhenReadParsed()) this[kOnReadParsed] = resumePausedUpgradeIncoming;
-        else resumePausedUpgradeIncoming(this);
+        // Paused: Node's UpgradeStream._read resumes the request only while its body is incomplete.
+        else if (response && (response.hasBody & NodeHTTPBodyReadState.done) !== 0) this[kUpgradeIncoming] = undefined;
+        else upgradeIncoming.resume();
         return;
       }
       response?.resume();
@@ -3169,6 +3194,7 @@ const DISPATCH_HAS_EXPECT = 1 << 4;
 const DISPATCH_EXPECT_CONTINUE = 1 << 5;
 const DISPATCH_HAS_CONTENT_LENGTH = 1 << 6;
 const DISPATCH_HAS_TRANSFER_ENCODING = 1 << 7;
+const DISPATCH_READ_HAS_MORE = 1 << 8;
 
 // Whether the response should advertise a persistent connection.
 // `connection` is the request's Connection header value (or undefined).

src/jsc/bindings/NodeHTTP.cpp

@@ -44,6 +44,8 @@ static constexpr uint32_t kDispatchHasExpect = 1 << 4;
 static constexpr uint32_t kDispatchExpectContinue = 1 << 5;
 static constexpr uint32_t kDispatchHasContentLength = 1 << 6;
 static constexpr uint32_t kDispatchHasTransferEncoding = 1 << 7;
+// Bytes follow the request head in the read that carried it.
+static constexpr uint32_t kDispatchReadHasMore = 1 << 8;
 
 static bool svEqualsIgnoreCase(std::string_view a, std::string_view lower)
 {
@@ -100,7 +102,7 @@ static void assignHeadersFromUWebSocketsForCall(uWS::HttpRequest* request, JSVal
         args.append(methodString);
     }
 
-    uint32_t bits = 0;
+    uint32_t bits = request->head.empty() ? 0 : kDispatchReadHasMore;
     // llhttp's F_CONNECTION_CLOSE / F_CONNECTION_UPGRADE: a whole list item.
     if (request->hasConnectionClose(true))
         bits |= kDispatchConnClose;