node:http: emit 'connect' for a CONNECT that is pipelined behind a pending response

src/js/internal/http.ts

@@ -39,6 +39,8 @@ const kInternalSocketData = Symbol.for("::bunternal::");
 const serverSymbol = Symbol.for("::bunternal::");
 const kPendingCallbacks = Symbol("pendingCallbacks");
 const kRequest = Symbol("request");
+// Set on a server socket at the 'connect'/'upgrade' handoff: the native response of that request.
+const kHandoffResponse = Symbol("kHandoffResponse");
 const kCloseCallback = Symbol("closeCallback");
 
 // node:_http_server registers its pipelined-response machinery here at module
@@ -521,6 +523,7 @@ export {
   kAbortController,
   kCloseCallback,
   kHandle,
+  kHandoffResponse,
   kInternalSocketData,
   kNeedDrain,
   kOutHeaders,

src/js/node/_http_incoming.ts

@@ -11,6 +11,7 @@ const {
   abortedSymbol,
   eofInProgress,
   kHandle,
+  kHandoffResponse,
   noBodySymbol,
   typeSymbol,
   NodeHTTPIncomingRequestType,
@@ -365,12 +366,13 @@ IncomingMessage.prototype._read = function _read(_n) {
   // Native server path.
   const socket = this.socket;
   if (socket && socket.readable) {
-    if (this.upgrade) {
+    if (this.upgrade || socket[kHandoffResponse] !== undefined) {
       // Upgrade request with a body (Node 26 semantics): reading the request
       // must not flip the raw socket into flowing mode - tunnel bytes pushed
       // to the socket before the 'upgrade' listener attaches its own 'data'
       // handler would be discarded by a flowing stream with no readers.
       // Resume the native body source directly instead.
+      // Same for a request ahead of a pipelined CONNECT: res.end() dumps it after the handoff.
       onIncomingMessageResumeNodeHTTPResponse.$call(this);
     } else {
       socket.resume();

src/js/node/_http_server.ts

@@ -32,6 +32,7 @@ const {
   kInternalSocketData,
   serverSymbol,
   kHandle,
+  kHandoffResponse,
   kRealListen,
   tlsSymbol,
   optionsSymbol,
@@ -735,7 +736,8 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
         socketParser[kParserOnTimeout] = serverParserShimOnTimeout;
 
         const isPipelined = !!isPipelinedDispatch;
-        if (method === "CONNECT" && !isPipelined) {
+        // Pipelined or not, like Node.js: the native parser is in tunnel mode from this request on.
+        if (method === "CONNECT") {
           // Handle CONNECT method for HTTP tunneling/proxy
           if (server.listenerCount("connect") > 0) {
             // For CONNECT, emit the event and let the handler respond
@@ -745,22 +747,30 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
             // The connection already parses as a CONNECT tunnel natively; this
             // additionally marks it half-open-capable so a client FIN ends the
             // readable side without tearing the tunnel down (allowHalfOpen).
-            socketHandle.upgradeToTunnel();
+            socketHandle.upgradeToTunnel(false, handle);
+            socket[kHandoffResponse] = handle;
             // The parser is detached: the socket is handed over with only
             // net.Socket's 'end' listener left, like Node.js.
             detachSocketListenersForHandoff(socket);
-            const promise = $newPromise();
             // Pass the pipelined data (head buffer) if any was received with the CONNECT request
             const head = connectHead ? connectHead : kEmptyBuffer;
             // Node.js's parserOnIncoming: req.upgrade is true for CONNECT
             // regardless of shouldUpgradeCallback.
             http_req.upgrade = true;
             // Node frees the parser before handing the raw socket to 'connect'.
             releaseServerParserShim(socket, http_req);
-            server.emit("connect", http_req, socket, head);
+            try {
+              server.emit("connect", http_req, socket, head);
+            } catch (err) {
+              // Like 'upgrade' below. Native would end the pending response, here the one ahead.
+              process.nextTick(rethrowUncaught, err);
+            }
+            // Native would tie a returned promise to the socket's current response, the one in flight.
+            if (isPipelined) return;
             // Attach the internal close listener after the user's "connect"
             // handler ran: Node.js hands the socket over with no listeners and
             // tests assert socket.listenerCount("close") === 0 there.
+            const promise = $newPromise();
             socket.once("close", resolveHandoffPromise.bind(undefined, promise));
             return promise;
           } else {
@@ -873,6 +883,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
         // token; the server then consults shouldUpgradeCallback (default: an
         // 'upgrade' listener is installed) and otherwise dispatches the
         // request normally.
+        // Not when pipelined: the builtin ws answers through the socket's current response, the one in flight.
         let is_upgrade = false;
         if (
           !isPipelined &&
@@ -962,7 +973,8 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           // 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.
-          socketHandle.upgradeToTunnel(hasBody);
+          socketHandle.upgradeToTunnel(hasBody, handle);
+          socket[kHandoffResponse] = handle;
           socket[kEnableStreaming](true);
           detachSocketListenersForHandoff(socket);
           // Node frees the parser before emitting 'upgrade' (socket.parser === null there).
@@ -1485,6 +1497,7 @@ function getNodeHTTPServerSocket() {
     [kBytesWritten] = 0;
     [kHandle];
     [kUpgradeIncoming] = undefined;
+    [kHandoffResponse] = undefined;
     server: Server;
     _httpMessage;
     _secureEstablished = false;
@@ -1585,7 +1598,8 @@ function getNodeHTTPServerSocket() {
           // detached socket stops reading and no longer keeps the process alive -
           // the never-used response for this request must not keep the event loop
           // alive either.
-          handle.response?.unref();
+          // Not handle.response: behind a pipelined CONNECT that is the response in flight.
+          this[kHandoffResponse]?.unref();
         }
 
         this.push(null);

src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp

@@ -104,6 +104,9 @@ void JSNodeHTTPServerSocket::close()
     }
 }
 
+template<bool SSL>
+static void onNodeHttpReadsResumable(us_socket_t* socket);
+
 template<bool SSL>
 static void upgradeToTunnelModeImpl(us_socket_t* socket, bool afterBody)
 {
@@ -113,12 +116,14 @@ static void upgradeToTunnelModeImpl(us_socket_t* socket, bool afterBody)
          * switch into tunnel mode once the message completes (Node 26 delivers
          * the body through the request before raw data starts flowing). */
         httpResponseData->state |= uWS::HttpResponseData<SSL>::HTTP_NODE_TUNNEL_AFTER_BODY;
-    } else {
-        httpResponseData->isConnectRequest = true;
+        return;
     }
+    httpResponseData->isConnectRequest = true;
+    /* resume() on a response does nothing in tunnel mode: lift what paused reads before it (req.pause(), flood prevention). */
+    onNodeHttpReadsResumable<SSL>(socket);
 }
 
-void JSNodeHTTPServerSocket::upgradeToTunnelMode(bool afterBody)
+void JSNodeHTTPServerSocket::upgradeToTunnelMode(bool afterBody, WebCore::JSNodeHTTPResponse* response)
 {
     if (!socket || us_socket_is_closed(socket)) {
         return;
@@ -136,9 +141,10 @@ void JSNodeHTTPServerSocket::upgradeToTunnelMode(bool afterBody)
         upgradeToTunnelModeImpl<false>(socket, afterBody);
     }
     /* The exchange leaves HTTP here: let the response release the server's
-     * pending-request accounting (see Flags::TUNNELED in NodeHTTPResponse.rs). */
-    if (auto* res = currentResponseObject.get(); res != nullptr && res->m_ctx != nullptr) {
-        Bun__NodeHTTPResponse_markTunneled(res->m_ctx);
+     * pending-request accounting (see Flags::TUNNELED in NodeHTTPResponse.rs).
+     * Not currentResponseObject: a pipelined request is queued instead. */
+    if (response != nullptr && response->m_ctx != nullptr) {
+        Bun__NodeHTTPResponse_markTunneled(response->m_ctx);
     }
 }
 
@@ -421,6 +427,14 @@ void JSNodeHTTPServerSocket::appendPipelinedResponse(JSC::VM& vm, WebCore::JSNod
     m_pipelinedResponses.last().set(vm, this, response);
 }
 
+/* A pipelined CONNECT stays queued for good, which keeps markDone() from marking the connection idle.
+ * No request follows it, so it holds no reads. */
+template<bool SSL>
+static bool queuedResponsesHoldReads(uWS::NodeHttpResponseData<SSL>* httpResponseData)
+{
+    return httpResponseData->nodeHttpQueuedPipelinedCount > 0 && !httpResponseData->isConnectRequest;
+}
+
 /* node:http flood prevention, resume half. Parked pipelined requests (HttpParser::nodeHttpPausedSpill)
  * must replay before fresh reads (ordering) and not synchronously inside the resuming JS operation.
  * Deferred as an event-loop task rooting the JS socket; reads resume once the spill drains without re-pausing. */
@@ -449,7 +463,7 @@ static void replayNodeHttpPausedSpill(us_socket_t* socket)
          * the rest; stay paused until the next resumable event. */
         return;
     }
-    if (httpResponseData->nodeHttpQueuedPipelinedCount > 0
+    if (queuedResponsesHoldReads<SSL>(httpResponseData)
         || reinterpret_cast<uWS::AsyncSocket<SSL>*>(socket)->getBufferedAmount() > 0) {
         /* The spill drained, but the pipeline has not: keep raw reads paused —
          * the queue-drain / writable events re-enter the hook. */
@@ -471,7 +485,7 @@ static void onNodeHttpReadsResumable(us_socket_t* socket)
             return;
         }
         if (httpResponseData->nodeHttpPausedSpill.isEmpty()
-            && httpResponseData->nodeHttpQueuedPipelinedCount > 0) {
+            && queuedResponsesHoldReads<SSL>(httpResponseData)) {
             return;
         }
     }

src/jsc/bindings/node/JSNodeHTTPServerSocket.h

@@ -113,7 +113,7 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject {
      * ondata callback as opaque data. With afterBody, the switch is deferred
      * until the request body has been fully parsed (Upgrade requests with a
      * body deliver it through the request first, like Node 26). */
-    void upgradeToTunnelMode(bool afterBody = false);
+    void upgradeToTunnelMode(bool afterBody, WebCore::JSNodeHTTPResponse* response);
 
     /* Trailer fields received after the current request's chunked body, as a
      * flat [name, value, ...] JS array preserving wire casing; jsUndefined()

src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp

@@ -118,9 +118,9 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketUpgradeToTunnel, (JSC::JS
     if (!thisObject) [[unlikely]] {
         return JSValue::encode(JSC::jsUndefined());
     }
-    // upgradeToTunnel(afterBody): with a truthy argument the switch happens only
+    // upgradeToTunnel(afterBody, response): with a truthy afterBody the switch happens only
     // once the request body has been fully parsed (Upgrade requests with a body).
-    thisObject->upgradeToTunnelMode(callFrame->argument(0).toBoolean(globalObject));
+    thisObject->upgradeToTunnelMode(callFrame->argument(0).toBoolean(globalObject), dynamicDowncast<WebCore::JSNodeHTTPResponse>(callFrame->argument(1)));
     return JSValue::encode(JSC::jsUndefined());
 }