fix(node:http): hand off upgrade socket to userland for bidirectional communication

packages/bun-uws/src/HttpResponse.h

@@ -764,6 +764,14 @@ struct HttpResponse : public AsyncSocket<SSL> {
         return httpResponseData->isConnectRequest;
     }
 
+    /* Mark this connection as a raw/tunnel connection (like CONNECT).
+     * This stops the HTTP parser from parsing subsequent data as HTTP,
+     * enabling raw bidirectional communication after protocol upgrade. */
+    void markAsRawMode() {
+        HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
+        httpResponseData->isConnectRequest = true;
+    }
+
     void setWriteOffset(uint64_t offset) {
         HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
 

src/bun.js/bindings/node/JSNodeHTTPServerSocketPrototype.cpp

@@ -6,6 +6,7 @@
 #include "helpers.h"
 #include <JavaScriptCore/JSCJSValueInlines.h>
 #include <wtf/text/WTFString.h>
+#include <bun-uws/src/App.h>
 
 extern "C" EncodedJSValue us_socket_buffered_js_write(void* socket, bool is_ssl, bool ended, us_socket_stream_buffer_t* streamBuffer, JSC::JSGlobalObject* globalObject, JSC::EncodedJSValue data, JSC::EncodedJSValue encoding);
 extern "C" uint64_t uws_res_get_remote_address_info(void* res, const char** dest, int* port, bool* is_ipv6);
@@ -28,6 +29,7 @@ JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterBytesWritten);
 JSC_DECLARE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketClose);
 JSC_DECLARE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketWrite);
 JSC_DECLARE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketEnd);
+JSC_DECLARE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketMarkAsRawMode);
 JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterResponse);
 JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterRemoteAddress);
 JSC_DECLARE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterLocalAddress);
@@ -56,6 +58,7 @@ static const JSC::HashTableValue JSNodeHTTPServerSocketPrototypeTableValues[] =
     { "close"_s, static_cast<unsigned>(JSC::PropertyAttribute::Function | JSC::PropertyAttribute::DontEnum), JSC::NoIntrinsic, { JSC::HashTableValue::NativeFunctionType, jsFunctionNodeHTTPServerSocketClose, 0 } },
     { "write"_s, static_cast<unsigned>(JSC::PropertyAttribute::Function | JSC::PropertyAttribute::DontEnum), JSC::NoIntrinsic, { JSC::HashTableValue::NativeFunctionType, jsFunctionNodeHTTPServerSocketWrite, 2 } },
     { "end"_s, static_cast<unsigned>(JSC::PropertyAttribute::Function | JSC::PropertyAttribute::DontEnum), JSC::NoIntrinsic, { JSC::HashTableValue::NativeFunctionType, jsFunctionNodeHTTPServerSocketEnd, 0 } },
+    { "markAsRawMode"_s, static_cast<unsigned>(JSC::PropertyAttribute::Function | JSC::PropertyAttribute::DontEnum), JSC::NoIntrinsic, { JSC::HashTableValue::NativeFunctionType, jsFunctionNodeHTTPServerSocketMarkAsRawMode, 0 } },
     { "secureEstablished"_s, static_cast<unsigned>(JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::ReadOnly), JSC::NoIntrinsic, { JSC::HashTableValue::GetterSetterType, jsNodeHttpServerSocketGetterIsSecureEstablished, noOpSetter } },
 };
 
@@ -113,6 +116,27 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketEnd, (JSC::JSGlobalObject
     return JSValue::encode(JSC::jsUndefined());
 }
 
+JSC_DEFINE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketMarkAsRawMode, (JSC::JSGlobalObject * globalObject, JSC::CallFrame* callFrame))
+{
+    auto* thisObject = jsDynamicCast<JSNodeHTTPServerSocket*>(callFrame->thisValue());
+    if (!thisObject) [[unlikely]] {
+        return JSValue::encode(JSC::jsUndefined());
+    }
+    if (thisObject->isClosed()) {
+        return JSValue::encode(JSC::jsUndefined());
+    }
+    // Set isConnectRequest on the HttpResponseData so that the HTTP parser
+    // stops parsing subsequent data as HTTP and passes it through as raw data.
+    if (thisObject->is_ssl) {
+        auto* httpResponseData = reinterpret_cast<uWS::HttpResponseData<true>*>(us_socket_ext(true, thisObject->socket));
+        httpResponseData->isConnectRequest = true;
+    } else {
+        auto* httpResponseData = reinterpret_cast<uWS::HttpResponseData<false>*>(us_socket_ext(false, thisObject->socket));
+        httpResponseData->isConnectRequest = true;
+    }
+    return JSValue::encode(JSC::jsUndefined());
+}
+
 // Implementation of custom getters
 JSC_DEFINE_CUSTOM_GETTER(jsNodeHttpServerSocketGetterIsSecureEstablished, (JSC::JSGlobalObject * globalObject, JSC::EncodedJSValue thisValue, JSC::PropertyName))
 {

src/deps/libuwsockets.cpp

@@ -1842,6 +1842,16 @@ __attribute__((callback (corker, ctx)))
       return uwsRes->isConnectRequest();
     }
   }
+  void uws_res_mark_as_raw_mode(int ssl, uws_res_r res)
+  {
+    if (ssl) {
+      uWS::HttpResponse<true> *uwsRes = (uWS::HttpResponse<true> *)res;
+      uwsRes->markAsRawMode();
+    } else {
+      uWS::HttpResponse<false> *uwsRes = (uWS::HttpResponse<false> *)res;
+      uwsRes->markAsRawMode();
+    }
+  }
   void *uws_res_get_native_handle(int ssl, uws_res_r res)
   {
     if (ssl)

src/deps/uws/Response.zig

@@ -42,6 +42,12 @@ pub fn NewResponse(ssl_flag: i32) type {
             return c.uws_res_is_connect_request(ssl_flag, res.downcast());
         }
 
+        /// Mark this connection as raw/tunnel mode (like CONNECT).
+        /// Stops the HTTP parser from parsing subsequent data as HTTP.
+        pub fn markAsRawMode(res: *Response) void {
+            c.uws_res_mark_as_raw_mode(ssl_flag, res.downcast());
+        }
+
         pub fn flushHeaders(res: *Response, flushImmediately: bool) void {
             c.uws_res_flush_headers(ssl_flag, res.downcast(), flushImmediately);
         }
@@ -590,6 +596,12 @@ pub const AnyResponse = union(enum) {
         };
     }
 
+    pub fn markAsRawMode(this: AnyResponse) void {
+        switch (this) {
+            inline else => |resp| resp.markAsRawMode(),
+        }
+    }
+
     pub fn endStream(this: AnyResponse, close_connection: bool) void {
         switch (this) {
             inline else => |resp| resp.endStream(close_connection),
@@ -672,6 +684,7 @@ const c = struct {
     pub extern fn us_socket_mark_needs_more_not_ssl(socket: ?*c.uws_res) void;
     pub extern fn uws_res_state(ssl: c_int, res: *const c.uws_res) State;
     pub extern fn uws_res_is_connect_request(ssl: i32, res: *c.uws_res) bool;
+    pub extern fn uws_res_mark_as_raw_mode(ssl: i32, res: *c.uws_res) void;
     pub extern fn uws_res_get_remote_address_info(res: *c.uws_res, dest: *[*]const u8, port: *i32, is_ipv6: *bool) usize;
     pub extern fn uws_res_uncork(ssl: i32, res: *c.uws_res) void;
     pub extern fn uws_res_end(ssl: i32, res: *c.uws_res, data: [*c]const u8, length: usize, close_connection: bool) void;

src/js/node/_http_server.ts

@@ -606,15 +606,16 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
           http_res.end();
           socket.destroy();
         } else if (is_upgrade) {
+          // Hand off the connection to userland, mirroring CONNECT handler.
+          // Enable streaming so socket.write() and socket.on("data") work.
+          socket[kEnableStreaming](true);
+          // Tell uWebSockets to stop parsing HTTP on this connection,
+          // switching to raw pass-through mode for bidirectional data.
+          socketHandle.markAsRawMode();
+          const { promise: upgradePromise, resolve: upgradeResolve } = $newPromiseCapability(Promise);
+          socket.once("close", upgradeResolve);
           server.emit("upgrade", http_req, socket, kEmptyBuffer);
-          if (!socket._httpMessage) {
-            if (canUseInternalAssignSocket) {
-              // ~10% performance improvement in JavaScriptCore due to avoiding .once("close", ...) and removing a listener
-              assignSocketInternal(http_res, socket);
-            } else {
-              http_res.assignSocket(socket);
-            }
-          }
+          return upgradePromise;
         } else if (http_req.headers.expect !== undefined) {
           if (http_req.headers.expect === "100-continue") {
             if (server.listenerCount("checkContinue") > 0) {

test/regression/issue/28157.test.ts

@@ -0,0 +1,78 @@
+import { expect, test } from "bun:test";
+import { bunEnv, bunExe } from "harness";
+
+test("node:http upgrade socket hands off to userland for bidirectional communication", async () => {
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+import http from "node:http";
+import net from "node:net";
+
+const server = http.createServer();
+
+server.on("upgrade", (req, socket, head) => {
+  socket.write(
+    "HTTP/1.1 101 Switching Protocols\\r\\n" +
+    "Upgrade: custom\\r\\n" +
+    "Connection: Upgrade\\r\\n" +
+    "\\r\\n"
+  );
+
+  socket.on("data", (chunk) => {
+    socket.write("ECHO:" + chunk.toString());
+  });
+
+  socket.resume();
+});
+
+server.listen(0, "127.0.0.1", () => {
+  const port = server.address().port;
+
+  const client = net.connect(port, "127.0.0.1", () => {
+    client.write(
+      "GET / HTTP/1.1\\r\\n" +
+      "Host: 127.0.0.1\\r\\n" +
+      "Upgrade: custom-protocol\\r\\n" +
+      "Connection: Upgrade\\r\\n" +
+      "\\r\\n"
+    );
+  });
+
+  let gotUpgrade = false;
+  let buf = "";
+
+  client.on("data", (chunk) => {
+    buf += chunk.toString();
+
+    if (!gotUpgrade && buf.includes("\\r\\n\\r\\n")) {
+      gotUpgrade = true;
+      client.write("hello from client");
+    }
+
+    if (buf.includes("ECHO:")) {
+      console.log(buf.substring(buf.indexOf("ECHO:")));
+      client.end();
+      server.close(() => process.exit(0));
+    }
+  });
+
+  setTimeout(() => {
+    console.error("TIMEOUT");
+    process.exit(1);
+  }, 5000);
+});
+`,
+    ],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+  expect(stderr).not.toContain("TIMEOUT");
+  expect(stdout.trim()).toBe("ECHO:hello from client");
+  expect(exitCode).toBe(0);
+});