fix node:http proxy support: createConnection, upgrade sockets, socket cleanup, connection close

packages/bun-uws/src/HttpResponse.h

@@ -154,6 +154,19 @@ struct HttpResponse : public AsyncSocket<SSL> {
                 }
             } else {
                 this->uncork();
+                /* After uncorking, check if we should close this connection.
+                 * This handles the case where writeHead corked the socket outside
+                 * the uWS request handler (async response), so the close check
+                 * in HttpContext's onData handler never runs for this response. */
+                if (httpResponseData->state & HttpResponseData<SSL>::HTTP_CONNECTION_CLOSE) {
+                    if ((httpResponseData->state & HttpResponseData<SSL>::HTTP_RESPONSE_PENDING) == 0) {
+                        if (((AsyncSocket<SSL> *) this)->getBufferedAmount() == 0) {
+                            ((AsyncSocket<SSL> *) this)->shutdown();
+                            ((AsyncSocket<SSL> *) this)->close();
+                            return true;
+                        }
+                    }
+                }
             }
 
             /* tryEnd can never fail when in chunked mode, since we do not have tryWrite (yet), only write */
@@ -219,6 +232,17 @@ struct HttpResponse : public AsyncSocket<SSL> {
                     }
                 }  else {
                     this->uncork();
+                    /* After uncorking, check if we should close this connection.
+                     * Same fix as the chunked path above. */
+                    if (httpResponseData->state & HttpResponseData<SSL>::HTTP_CONNECTION_CLOSE) {
+                        if ((httpResponseData->state & HttpResponseData<SSL>::HTTP_RESPONSE_PENDING) == 0) {
+                            if (((AsyncSocket<SSL> *) this)->getBufferedAmount() == 0) {
+                                ((AsyncSocket<SSL> *) this)->shutdown();
+                                ((AsyncSocket<SSL> *) this)->close();
+                                return true;
+                            }
+                        }
+                    }
                 }
             }
 
@@ -764,6 +788,15 @@ struct HttpResponse : public AsyncSocket<SSL> {
         return httpResponseData->isConnectRequest;
     }
 
+    /* Mark this response as a connect-like request so that subsequent data
+     * is forwarded as raw bytes instead of going through the HTTP parser.
+     * Used for HTTP upgrade (e.g. WebSocket) when the application takes
+     * ownership of the socket via the node:http upgrade event. */
+    void markAsConnectRequest() {
+        HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
+        httpResponseData->isConnectRequest = true;
+    }
+
     void setWriteOffset(uint64_t offset) {
         HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
 

src/bun.js/api/server.classes.ts

@@ -155,6 +155,10 @@ export default [
         fn: "doResume",
         length: 0,
       },
+      markAsUpgraded: {
+        fn: "markAsUpgraded",
+        length: 0,
+      },
       bufferedAmount: {
         getter: "getBufferedAmount",
       },

src/bun.js/api/server/NodeHTTPResponse.zig

@@ -618,6 +618,17 @@ pub fn doResume(this: *NodeHTTPResponse, globalObject: *jsc.JSGlobalObject, _: *
     return result;
 }
 
+pub fn markAsUpgraded(this: *NodeHTTPResponse, _: *jsc.JSGlobalObject, _: *jsc.CallFrame) jsc.JSValue {
+    log("markAsUpgraded", .{});
+    if (this.raw_response) |resp| {
+        // Mark the uWS response as a connect request so that subsequent data
+        // is forwarded as raw bytes instead of going through the HTTP parser.
+        // This enables bidirectional streaming after an HTTP upgrade.
+        resp.markAsConnectRequest();
+    }
+    return .js_undefined;
+}
+
 pub fn onRequestComplete(this: *NodeHTTPResponse) void {
     if (this.flags.request_has_completed) {
         return;

src/deps/libuwsockets.cpp

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

src/deps/uws/Response.zig

@@ -42,6 +42,10 @@ pub fn NewResponse(ssl_flag: i32) type {
             return c.uws_res_is_connect_request(ssl_flag, res.downcast());
         }
 
+        pub fn markAsConnectRequest(res: *Response) void {
+            c.uws_res_mark_as_connect_request(ssl_flag, res.downcast());
+        }
+
         pub fn flushHeaders(res: *Response, flushImmediately: bool) void {
             c.uws_res_flush_headers(ssl_flag, res.downcast(), flushImmediately);
         }
@@ -590,6 +594,12 @@ pub const AnyResponse = union(enum) {
         };
     }
 
+    pub fn markAsConnectRequest(this: AnyResponse) void {
+        switch (this) {
+            inline else => |resp| resp.markAsConnectRequest(),
+        }
+    }
+
     pub fn endStream(this: AnyResponse, close_connection: bool) void {
         switch (this) {
             inline else => |resp| resp.endStream(close_connection),
@@ -672,6 +682,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_connect_request(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;