Merge remote-tracking branch 'origin/main' into ciro/fetch-upgrade-node-http

Refactor the ClientRequest constructor to eliminate anonymous function
closures that capture the entire constructor scope. This improves memory
characteristics by allowing earlier garbage collection.

Changes:
- Add symbols to internal/http.ts for mutable state: kWriteCount,
  kResolveNextChunk, kFetching, kOnEnd, kHandleResponse
- Move emit helpers to module scope: maybeEmitSocketClientRequest,
  maybeEmitPrefinishClientRequest, maybeEmitFinishClientRequest,
  maybeEmitCloseClientRequest
- Move event handlers to module scope: socketCloseListenerClientRequest,
  onAbortClientRequest
- Move write functions to module scope: pushChunkClientRequest,
  writeInternalClientRequest, writeClientRequest, endClientRequest
- Move utility functions to module scope: flushHeadersClientRequest,
  destroyClientRequest, abortClientRequest, ensureTlsClientRequest,
  signalAbortHandler, clearTimeoutClientRequest
- Move send/fetch functions to module scope: sendClientRequest,
  bodyIteratorClientRequest, goClientRequest, startFetchClientRequest,
  iterateCandidatesClientRequest, getURLClientRequest,
  failDNSLookupClientRequest

The streaming body generator still uses a minimal closure that only
captures 'self', with all mutable state stored in symbol properties.

src/js/internal/http.ts

@@ -57,6 +57,11 @@ const kSignal = Symbol("signal");
 const kMaxHeaderSize = Symbol("maxHeaderSize");
 const abortedSymbol = Symbol("aborted");
 const kClearTimeout = Symbol("kClearTimeout");
+const kWriteCount = Symbol("writeCount");
+const kResolveNextChunk = Symbol("resolveNextChunk");
+const kFetching = Symbol("fetching");
+const kOnEnd = Symbol("onEnd");
+const kHandleResponse = Symbol("handleResponse");
 
 const headerStateSymbol = Symbol("headerState");
 // used for pretending to emit events in the right order
@@ -353,6 +358,7 @@ function emitErrorNt(msg, err, callback) {
 const setMaxHTTPHeaderSize = $newZigFunction("node_http_binding.zig", "setMaxHTTPHeaderSize", 1);
 const getMaxHTTPHeaderSize = $newZigFunction("node_http_binding.zig", "getMaxHTTPHeaderSize", 0);
 const kOutHeaders = Symbol("kOutHeaders");
+const kEmptyBuffer = Buffer.alloc(0);
 
 function ipToInt(ip) {
   const octets = ip.split(".");
@@ -517,14 +523,18 @@ export {
   kDeferredTimeouts,
   kDeprecatedReplySymbol,
   kEmitState,
+  kEmptyBuffer,
   kEmptyObject,
   kFetchRequest,
+  kFetching,
   kHandle,
+  kHandleResponse,
   kHost,
   kInternalSocketData,
   kMaxHeaderSize,
   kMaxHeadersCount,
   kMethod,
+  kOnEnd,
   kOptions,
   kOutHeaders,
   kParser,
@@ -535,6 +545,7 @@ export {
   kRealListen,
   kRequest,
   kRes,
+  kResolveNextChunk,
   kReusedSocket,
   kSignal,
   kSocketPath,
@@ -542,6 +553,7 @@ export {
   kTls,
   kUpgradeOrConnect,
   kUseDefaultPort,
+  kWriteCount,
   noBodySymbol,
   optionsSymbol,
   parseProxyConfigFromEnv,

src/js/node/_http_server.ts

@@ -43,6 +43,7 @@ const {
   setServerIdleTimeout,
   setServerCustomOptions,
   getMaxHTTPHeaderSize,
+  kEmptyBuffer,
 } = require("internal/http");
 const NumberIsNaN = Number.isNaN;
 
@@ -59,7 +60,7 @@ const sendHelper = $newZigFunction("node_cluster_binding.zig", "sendHelperChild"
 const kServerResponse = Symbol("ServerResponse");
 const kRejectNonStandardBodyWrites = Symbol("kRejectNonStandardBodyWrites");
 const GlobalPromise = globalThis.Promise;
-const kEmptyBuffer = Buffer.alloc(0);
+
 const ObjectKeys = Object.keys;
 const MathMin = Math.min;
 

test/js/node/test/parallel/test-http-upgrade-client.js

@@ -0,0 +1,99 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+'use strict';
+// Verify that the 'upgrade' header causes an 'upgrade' event to be emitted to
+// the HTTP client. This test uses a raw TCP server to better control server
+// behavior.
+
+const common = require('../common');
+const assert = require('assert');
+
+const http = require('http');
+const net = require('net');
+const Countdown = require('../common/countdown');
+
+const expectedRecvData = 'nurtzo';
+
+// Create a TCP server
+const server = net.createServer(function(c) {
+  c.on('data', function(d) {
+    c.write('HTTP/1.1 101\r\n');
+    c.write('hello: world\r\n');
+    c.write('connection: upgrade\r\n');
+    c.write('upgrade: websocket\r\n');
+    c.write('\r\n');
+    c.write(expectedRecvData);
+  });
+
+  c.on('end', function() {
+    c.end();
+  });
+});
+
+server.listen(0, "127.0.0.1", common.mustCall(function() {
+  const port = this.address().port;
+  const headers = [
+    {
+      connection: 'upgrade',
+      upgrade: 'websocket'
+    },
+    [
+      ['Host', 'echo.websocket.org'],
+      ['Connection', 'Upgrade'],
+      ['Upgrade', 'websocket'],
+      ['Origin', 'http://www.websocket.org'],
+    ],
+  ];
+  const countdown = new Countdown(headers.length, () => server.close());
+
+  headers.forEach(function(h) {
+    const req = http.get({
+      port: port,
+      hostname: "127.0.0.1",
+      headers: h
+    });
+    let sawUpgrade = false;
+    req.on('upgrade', common.mustCall(function(res, socket, upgradeHead) {
+      sawUpgrade = true;
+      let recvData = upgradeHead;
+      socket.on('data', function(d) {
+        recvData += d;
+      });
+
+      socket.on('close', common.mustCall(function() {
+        assert.strictEqual(recvData.toString(), expectedRecvData);
+      }));
+
+      const expectedHeaders = {
+        hello: 'world',
+        connection: 'upgrade',
+        upgrade: 'websocket'
+      };
+      assert.deepStrictEqual(res.headers, expectedHeaders);
+      socket.end();
+      countdown.dec();
+    }));
+    req.on('close', common.mustCall(function() {
+      assert.strictEqual(sawUpgrade, true);
+    }));
+  });
+}));

test/js/web/fetch/fetch.upgrade.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, test } from "bun:test";
+import http from "http";
 import { decodeFrames, encodeCloseFrame, encodeTextFrame, upgradeHeaders } from "./websocket.helpers";
 
 describe("fetch upgrade", () => {
@@ -60,4 +61,80 @@ describe("fetch upgrade", () => {
     expect(serverMessages).toEqual(["hello", "world", "bye", "close"]);
     expect(clientMessages).toEqual(["Hello World", "close"]);
   });
+
+  test("should upgrade to websocket using http.request", async () => {
+    const serverMessages: string[] = [];
+    using server = Bun.serve({
+      port: 0,
+      fetch(req) {
+        if (server.upgrade(req)) return;
+        return new Response("Hello World");
+      },
+      websocket: {
+        open(ws) {
+          ws.send("Hello World");
+        },
+        message(ws, message) {
+          serverMessages.push(message as string);
+        },
+        close(ws) {
+          serverMessages.push("close");
+        },
+      },
+    });
+    const req = http.request(
+      {
+        port: server.url.port,
+        hostname: server.url.hostname,
+        headers: upgradeHeaders(),
+      },
+      res => {
+        expect.unreachable("should not call response callback");
+      },
+    );
+    const clientMessages: string[] = [];
+
+    const { promise, resolve, reject } = Promise.withResolvers<void>();
+
+    req.on("upgrade", (req, socket, head) => {
+      try {
+        expect(req.statusCode).toBe(101);
+        expect(req.headers.upgrade).toBe("websocket");
+        expect(req.headers["sec-websocket-accept"]).toBeDefined();
+        expect(req.headers.connection).toBe("Upgrade");
+        expect(head).toBeDefined();
+        expect(Buffer.isBuffer(head)).toBe(true);
+        function onData(data: Buffer) {
+          for (const msg of decodeFrames(data)) {
+            if (typeof msg === "string") {
+              clientMessages.push(msg);
+            } else {
+              clientMessages.push(msg.type);
+              if (msg.type === "close") {
+                socket.end();
+                resolve();
+              }
+            }
+          }
+        }
+        if (head.length > 0) {
+          onData(head);
+        }
+        socket.on("data", onData);
+
+        socket.write(encodeTextFrame("hello"));
+        socket.write(encodeTextFrame("world"));
+        socket.write(encodeTextFrame("bye"));
+        socket.write(encodeCloseFrame());
+      } catch (err) {
+        reject(err);
+      }
+    });
+
+    req.end();
+    await promise;
+
+    expect(serverMessages).toEqual(["hello", "world", "bye", "close"]);
+    expect(clientMessages).toEqual(["Hello World", "close"]);
+  });
 });