fix(ext/node): tls.connect socket upgrades (#27125)

Fixes #27087
Fixes #26685
Fixes #26660

ext/node/polyfills/_tls_wrap.ts

@@ -148,9 +148,13 @@ export class TLSSocket extends net.Socket {
           : new TCP(TCPConstants.SOCKET);
       }
 
+      const { promise, resolve } = Promise.withResolvers();
+
       // Patches `afterConnect` hook to replace TCP conn with TLS conn
       const afterConnect = handle.afterConnect;
       handle.afterConnect = async (req: any, status: number) => {
+        options.hostname ??= undefined; // coerce to undefined if null, startTls expects hostname to be undefined
+
         try {
           const conn = await Deno.startTls(handle[kStreamBaseField], options);
           try {
@@ -164,15 +168,25 @@ export class TLSSocket extends net.Socket {
             // Don't interrupt "secure" event to let the first read/write
             // operation emit the error.
           }
+
+          // Assign the TLS connection to the handle and resume reading.
           handle[kStreamBaseField] = conn;
+          handle.upgrading = false;
+          if (!handle.pauseOnCreate) {
+            handle.readStart();
+          }
+
+          resolve();
+
           tlssock.emit("secure");
           tlssock.removeListener("end", onConnectEnd);
-        } catch (_) {
+        } catch {
           // TODO(kt3k): Handle this
         }
         return afterConnect.call(handle, req, status);
       };
 
+      handle.upgrading = promise;
       (handle as any).verifyError = function () {
         return null; // Never fails, rejectUnauthorized is always true in Deno.
       };

ext/node/polyfills/http2.ts

@@ -479,13 +479,13 @@ export class ClientHttp2Session extends Http2Session {
 
     socket.on("error", socketOnError);
     socket.on("close", socketOnClose);
+
+    socket[kHandle].pauseOnCreate = true;
     const connPromise = new Promise((resolve) => {
       const eventName = url.startsWith("https") ? "secureConnect" : "connect";
       socket.once(eventName, () => {
         const rid = socket[kHandle][kStreamBaseField][internalRidSymbol];
-        nextTick(() => {
-          resolve(rid);
-        });
+        nextTick(() => resolve(rid));
       });
     });
     socket[kSession] = this;

ext/node/polyfills/internal_binding/stream_wrap.ts

@@ -320,8 +320,16 @@ export class LibuvStreamWrap extends HandleWrap {
   /** Internal method for reading from the attached stream. */
   async #read() {
     let buf = this.#buf;
+
     let nread: number | null;
     const ridBefore = this[kStreamBaseField]![internalRidSymbol];
+
+    if (this.upgrading) {
+      // Starting an upgrade, stop reading. Upgrading will resume reading.
+      this.readStop();
+      return;
+    }
+
     try {
       nread = await this[kStreamBaseField]!.read(buf);
     } catch (e) {
@@ -382,6 +390,11 @@ export class LibuvStreamWrap extends HandleWrap {
 
     const ridBefore = this[kStreamBaseField]![internalRidSymbol];
 
+    if (this.upgrading) {
+      // There is an upgrade in progress, queue the write request.
+      await this.upgrading;
+    }
+
     let nwritten = 0;
     try {
       // TODO(crowlKats): duplicate from runtime/js/13_buffer.js
@@ -400,7 +413,6 @@ export class LibuvStreamWrap extends HandleWrap {
       }
 
       let status: number;
-
       // TODO(cmorten): map err to status codes
       if (
         e instanceof Deno.errors.BadResource ||

tests/unit_node/tls_test.ts

@@ -257,3 +257,17 @@ Deno.test("TLSSocket.alpnProtocol is set for client", async () => {
   listener.close();
   await new Promise((resolve) => outgoing.on("close", resolve));
 });
+
+Deno.test("tls connect upgrade tcp", async () => {
+  const { promise, resolve } = Promise.withResolvers<void>();
+
+  const socket = new net.Socket();
+  socket.connect(443, "google.com");
+  socket.on("connect", () => {
+    const secure = tls.connect({ socket });
+    secure.on("secureConnect", () => resolve());
+  });
+
+  await promise;
+  socket.destroy();
+});