fix(ext/http): internal upgradeHttpRaw works with "Deno.serve()" API (#…

…18859)

Fix internal "upgradeHttpRaw" API restoring capability to upgrade HTTP
connection in polyfilles "node:http" API.

cli/tests/unit/serve_test.ts

@@ -17,6 +17,11 @@ import {
 } from "./test_util.ts";
 import { consoleSize } from "../../../runtime/js/40_tty.js";
 
+const {
+  upgradeHttpRaw,
+  // @ts-expect-error TypeScript (as of 3.7) does not support indexing namespaces by symbol
+} = Deno[Deno.internal];
+
 function createOnErrorCb(ac: AbortController): (err: unknown) => Response {
   return (err) => {
     console.error(err);
@@ -803,6 +808,85 @@ Deno.test({ permissions: { net: true } }, async function httpServerWebSocket() {
   await server;
 });
 
+Deno.test(
+  { permissions: { net: true } },
+  async function httpServerWebSocketRaw() {
+    const ac = new AbortController();
+    const listeningPromise = deferred();
+    const server = Deno.serve({
+      handler: async (request) => {
+        const { conn, response } = upgradeHttpRaw(request);
+        const buf = new Uint8Array(1024);
+        let read;
+
+        // Write our fake HTTP upgrade
+        await conn.write(
+          new TextEncoder().encode(
+            "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgraded\r\n\r\nExtra",
+          ),
+        );
+
+        // Upgrade data
+        read = await conn.read(buf);
+        assertEquals(
+          new TextDecoder().decode(buf.subarray(0, read!)),
+          "Upgrade data",
+        );
+        // Read the packet to echo
+        read = await conn.read(buf);
+        // Echo
+        await conn.write(buf.subarray(0, read!));
+
+        conn.close();
+        return response;
+      },
+      port: 4501,
+      signal: ac.signal,
+      onListen: onListen(listeningPromise),
+      onError: createOnErrorCb(ac),
+    });
+
+    await listeningPromise;
+
+    const conn = await Deno.connect({ port: 4501 });
+    await conn.write(
+      new TextEncoder().encode(
+        "GET / HTTP/1.1\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\nUpgrade data",
+      ),
+    );
+    const buf = new Uint8Array(1024);
+    let len;
+
+    // Headers
+    let headers = "";
+    for (let i = 0; i < 2; i++) {
+      len = await conn.read(buf);
+      headers += new TextDecoder().decode(buf.subarray(0, len!));
+      if (headers.endsWith("Extra")) {
+        break;
+      }
+    }
+    assertMatch(
+      headers,
+      /HTTP\/1\.1 101 Switching Protocols[ ,.A-Za-z:0-9\r\n]*Extra/im,
+    );
+
+    // Data to echo
+    await conn.write(new TextEncoder().encode("buffer data"));
+
+    // Echo
+    len = await conn.read(buf);
+    assertEquals(
+      new TextDecoder().decode(buf.subarray(0, len!)),
+      "buffer data",
+    );
+
+    conn.close();
+    ac.abort();
+    await server;
+  },
+);
+
 Deno.test(
   { permissions: { net: true } },
   async function httpServerWebSocketUpgradeTwice() {

cli/tsc/dts/lib.deno.unstable.d.ts

@@ -1516,25 +1516,6 @@ declare namespace Deno {
     request: Request,
   ): Promise<[Deno.Conn, Uint8Array]>;
 
-  /** **UNSTABLE**: New API, yet to be vetted.
-   *
-   * Allows "hijacking" the connection that the request is associated with.
-   * This can be used to implement protocols that build on top of HTTP (eg.
-   * {@linkcode WebSocket}).
-   *
-   * Unlike {@linkcode Deno.upgradeHttp} this function does not require that you
-   * respond to the request with a {@linkcode Response} object. Instead this
-   * function returns the underlying connection and first packet received
-   * immediately, and then the caller is responsible for writing the response to
-   * the connection.
-   *
-   * This method can only be called on requests originating the
-   * {@linkcode Deno.serve} server.
-   *
-   * @category HTTP Server
-   */
-  export function upgradeHttpRaw(request: Request): [Deno.Conn, Uint8Array];
-
   /** **UNSTABLE**: New API, yet to be vetted.
    *
    * Open a new {@linkcode Deno.Kv} connection to persist data.

ext/http/00_serve.js

@@ -1,6 +1,7 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 const core = globalThis.Deno.core;
 const primordials = globalThis.__bootstrap.primordials;
+const internals = globalThis.__bootstrap.internals;
 
 const { BadResourcePrototype } = core;
 import { InnerBody } from "ext:deno_fetch/22_body.js";
@@ -10,7 +11,7 @@ import {
   newInnerResponse,
   toInnerResponse,
 } from "ext:deno_fetch/23_response.js";
-import { fromInnerRequest } from "ext:deno_fetch/23_request.js";
+import { fromInnerRequest, toInnerRequest } from "ext:deno_fetch/23_request.js";
 import { AbortController } from "ext:deno_web/03_abort_signal.js";
 import {
   _eventLoop,
@@ -32,6 +33,7 @@ import {
   readableStreamForRid,
   ReadableStreamPrototype,
 } from "ext:deno_web/06_streams.js";
+import { TcpConn } from "ext:deno_net/01_net.js";
 const {
   ObjectPrototypeIsPrototypeOf,
   SafeSet,
@@ -82,6 +84,14 @@ const UPGRADE_RESPONSE_SENTINEL = fromInnerResponse(
   "immutable",
 );
 
+function upgradeHttpRaw(req, conn) {
+  const inner = toInnerRequest(req);
+  if (inner._wantsUpgrade) {
+    return inner._wantsUpgrade("upgradeHttpRaw", conn);
+  }
+  throw new TypeError("upgradeHttpRaw may only be used with Deno.serve");
+}
+
 class InnerRequest {
   #slabId;
   #context;
@@ -122,10 +132,26 @@ class InnerRequest {
       throw "upgradeHttp is unavailable in Deno.serve at this time";
     }
 
-    // upgradeHttpRaw is async
-    // TODO(mmastrac)
+    // upgradeHttpRaw is sync
     if (upgradeType == "upgradeHttpRaw") {
-      throw "upgradeHttp is unavailable in Deno.serve at this time";
+      const slabId = this.#slabId;
+      const underlyingConn = originalArgs[0];
+
+      this.url();
+      this.headerList;
+      this.close();
+
+      this.#upgraded = () => {};
+
+      const upgradeRid = core.ops.op_upgrade_raw(slabId);
+
+      const conn = new TcpConn(
+        upgradeRid,
+        underlyingConn?.remoteAddr,
+        underlyingConn?.localAddr,
+      );
+
+      return { response: UPGRADE_RESPONSE_SENTINEL, conn };
     }
 
     // upgradeWebSocket is sync
@@ -623,4 +649,6 @@ async function serve(arg1, arg2) {
   }
 }
 
-export { serve };
+internals.upgradeHttpRaw = upgradeHttpRaw;
+
+export { serve, upgradeHttpRaw };

ext/http/01_http.js

@@ -64,7 +64,6 @@ const {
 } = primordials;
 
 const connErrorSymbol = Symbol("connError");
-const streamRid = Symbol("streamRid");
 const _deferred = Symbol("upgradeHttpDeferred");
 
 class HttpConn {
@@ -482,16 +481,6 @@ function upgradeHttp(req) {
   return req[_deferred].promise;
 }
 
-async function upgradeHttpRaw(req, tcpConn) {
-  const inner = toInnerRequest(req);
-  if (inner._wantsUpgrade) {
-    return inner._wantsUpgrade("upgradeHttpRaw", arguments);
-  }
-
-  const res = await core.opAsync("op_http_upgrade_early", inner[streamRid]);
-  return new TcpConn(res, tcpConn.remoteAddr, tcpConn.localAddr);
-}
-
 const spaceCharCode = StringPrototypeCharCodeAt(" ", 0);
 const tabCharCode = StringPrototypeCharCodeAt("\t", 0);
 const commaCharCode = StringPrototypeCharCodeAt(",", 0);
@@ -566,4 +555,4 @@ function buildCaseInsensitiveCommaValueFinder(checkText) {
 internals.buildCaseInsensitiveCommaValueFinder =
   buildCaseInsensitiveCommaValueFinder;
 
-export { _ws, HttpConn, serve, upgradeHttp, upgradeHttpRaw, upgradeWebSocket };
+export { _ws, HttpConn, serve, upgradeHttp, upgradeWebSocket };