refactor: "Deno.serve()" API uses "Deno.serveHttp()" internally (#18568)

This commit changes implementation of "Deno.serve()" API to use
"Deno.serveHttp()" under the hood. This change will allow us to
remove the "flash" server implementation, bringing stability to the
"Deno.serve()" API.

"cli/tests/unit/flash_test.ts" was renamed to "serve_test.ts".

Closes https://github.com/denoland/deno/issues/15574
Closes https://github.com/denoland/deno/issues/15504
Closes https://github.com/denoland/deno/issues/15646
Closes https://github.com/denoland/deno/issues/15909
Closes https://github.com/denoland/deno/issues/15911
Closes https://github.com/denoland/deno/issues/16828
Closes https://github.com/denoland/deno/issues/18046
Closes https://github.com/denoland/deno/issues/15869

cli/tests/integration/js_unit_tests.rs

@@ -26,8 +26,6 @@ fn js_unit_tests() {
     .current_dir(util::root_path())
     .arg("test")
     .arg("--unstable")
-    // Flash tests are crashing with SIGSEGV on Ubuntu, so we'll disable these entirely
-    .arg("--ignore=./cli/tests/unit/flash_test.ts")
     .arg("--location=http://js-unit-tests/foo/bar")
     .arg("--no-prompt")
     .arg("-A")

cli/tests/unit/serve_test.ts

@@ -301,51 +301,6 @@ Deno.test(
   },
 );
 
-Deno.test(
-  { permissions: { net: true } },
-  async function httpReadHeadersAfterClose() {
-    const promise = deferred();
-    const ac = new AbortController();
-    const listeningPromise = deferred();
-
-    let req: Request;
-    const server = Deno.serve({
-      handler: async (request) => {
-        await request.text();
-        req = request;
-        promise.resolve();
-        return new Response("Hello World");
-      },
-      port: 2334,
-      signal: ac.signal,
-      onListen: onListen(listeningPromise),
-      onError: createOnErrorCb(ac),
-    });
-
-    await listeningPromise;
-    const conn = await Deno.connect({ port: 2334 });
-    // Send GET request with a body + content-length.
-    const encoder = new TextEncoder();
-    const body =
-      `GET / HTTP/1.1\r\nHost: 127.0.0.1:2333\r\nContent-Length: 5\r\n\r\n12345`;
-    const writeResult = await conn.write(encoder.encode(body));
-    assertEquals(body.length, writeResult);
-    await promise;
-    conn.close();
-
-    assertThrows(
-      () => {
-        req.headers;
-      },
-      TypeError,
-      "request closed",
-    );
-
-    ac.abort();
-    await server;
-  },
-);
-
 Deno.test(
   { permissions: { net: true } },
   async function httpServerGetRequestBody() {
@@ -505,8 +460,10 @@ Deno.test(
         const body = new ReadableStream({
           start(controller) {
             // Non-encoded string is not a valid readable chunk.
+            // @ts-ignore we're testing that input is invalid
             controller.enqueue("wat");
           },
+          type: "bytes",
         });
         return new Response(body);
       },
@@ -518,21 +475,16 @@ Deno.test(
           `Internal server error: ${(err as Error).message}`,
           { status: 500 },
         );
-        ac.abort();
-        errorPromise.resolve(errResp);
+        errorPromise.resolve();
         return errResp;
       },
     });
 
     await listeningPromise;
-
     const resp = await fetch("http://127.0.0.1:4501/");
     // Incorrectly implemented reader ReadableStream should reject.
-    await assertRejects(() => resp.body!.getReader().read());
-
-    const err = await errorPromise as Response;
-    assertStringIncludes(await err.text(), "Expected ArrayBufferView");
-
+    assertStringIncludes(await resp.text(), "Failed to execute 'enqueue'");
+    await errorPromise;
     ac.abort();
     await server;
   },
@@ -571,7 +523,7 @@ Deno.test(
 
     ac.abort();
     await server;
-    assert(msg.includes("Content-Length: 60"));
+    assert(msg.includes("content-length: 60"));
   },
 );
 
@@ -912,7 +864,7 @@ Deno.test(
     await clientConn.read(buf);
 
     await promise;
-    let responseText = new TextDecoder().decode(buf);
+    let responseText = new TextDecoder("iso-8859-1").decode(buf);
     clientConn.close();
 
     assert(/\r\n[Xx]-[Hh]eader-[Tt]est: Æ\r\n/.test(responseText));
@@ -986,7 +938,7 @@ Deno.test(
     const server = Deno.serve({
       handler: async (request) => {
         assertEquals(await request.text(), "");
-        assertEquals(request.headers.get("cookie"), "foo=bar, bar=foo");
+        assertEquals(request.headers.get("cookie"), "foo=bar; bar=foo");
         promise.resolve();
         return new Response("ok");
       },
@@ -1122,68 +1074,6 @@ Deno.test(
   },
 );
 
-Deno.test("upgradeHttpRaw tcp", async () => {
-  const promise = deferred();
-  const listeningPromise = deferred();
-  const promise2 = deferred();
-  const ac = new AbortController();
-  const signal = ac.signal;
-  let conn: Deno.Conn;
-  let _head;
-  const handler = (req: Request) => {
-    [conn, _head] = Deno.upgradeHttpRaw(req);
-
-    (async () => {
-      await conn.write(
-        new TextEncoder().encode("HTTP/1.1 101 Switching Protocols\r\n\r\n"),
-      );
-
-      promise.resolve();
-
-      const buf = new Uint8Array(1024);
-      const n = await conn.read(buf);
-
-      assert(n != null);
-      const secondPacketText = new TextDecoder().decode(buf.slice(0, n));
-      assertEquals(secondPacketText, "bla bla bla\nbla bla\nbla\n");
-
-      promise2.resolve();
-    })();
-  };
-  const server = Deno.serve({
-    // NOTE: `as any` is used to bypass type checking for the return value
-    // of the handler.
-    handler: handler as any,
-    port: 4501,
-    signal,
-    onListen: onListen(listeningPromise),
-    onError: createOnErrorCb(ac),
-  });
-
-  await listeningPromise;
-  const tcpConn = await Deno.connect({ port: 4501 });
-  await tcpConn.write(
-    new TextEncoder().encode(
-      "CONNECT server.example.com:80 HTTP/1.1\r\n\r\n",
-    ),
-  );
-
-  await promise;
-
-  await tcpConn.write(
-    new TextEncoder().encode(
-      "bla bla bla\nbla bla\nbla\n",
-    ),
-  );
-
-  await promise2;
-  conn!.close();
-  tcpConn.close();
-
-  ac.abort();
-  await server;
-});
-
 // Some of these tests are ported from Hyper
 // https://github.com/hyperium/hyper/blob/889fa2d87252108eb7668b8bf034ffcc30985117/src/proto/h1/role.rs
 // https://github.com/hyperium/hyper/blob/889fa2d87252108eb7668b8bf034ffcc30985117/tests/server.rs
@@ -1610,7 +1500,7 @@ Deno.test(
     const readResult = await conn.read(buf);
     assert(readResult);
     const msg = decoder.decode(buf.subarray(0, readResult));
-    assert(msg.endsWith("HTTP/1.1 400 Bad Request\r\n\r\n"));
+    assert(msg.includes("HTTP/1.1 400 Bad Request"));
 
     conn.close();
 
@@ -1727,7 +1617,7 @@ Deno.test(
     assert(readResult);
     const msg = decoder.decode(buf.subarray(0, readResult));
 
-    assert(msg.endsWith("Content-Length: 300\r\n\r\n"));
+    assert(msg.includes("content-length: 300\r\n"));
 
     conn.close();
 
@@ -1921,7 +1811,7 @@ Deno.test(
       const readResult = await conn.read(buf);
       assert(readResult);
       const msg = decoder.decode(buf.subarray(0, readResult));
-      assert(msg.endsWith("HTTP/1.1 400 Bad Request\r\n\r\n"));
+      assert(msg.includes("HTTP/1.1 400 Bad Request\r\n"));
 
       conn.close();
     }
@@ -2164,44 +2054,6 @@ for (const [name, req] of badRequests) {
   );
 }
 
-Deno.test(
-  { permissions: { net: true } },
-  async function httpServerImplicitZeroContentLengthForHead() {
-    const ac = new AbortController();
-    const listeningPromise = deferred();
-
-    const server = Deno.serve({
-      handler: () => new Response(null),
-      port: 4503,
-      signal: ac.signal,
-      onListen: onListen(listeningPromise),
-      onError: createOnErrorCb(ac),
-    });
-
-    await listeningPromise;
-    const conn = await Deno.connect({ port: 4503 });
-    const encoder = new TextEncoder();
-    const decoder = new TextDecoder();
-
-    const body =
-      `HEAD / HTTP/1.1\r\nHost: example.domain\r\nConnection: close\r\n\r\n`;
-    const writeResult = await conn.write(encoder.encode(body));
-    assertEquals(body.length, writeResult);
-
-    const buf = new Uint8Array(1024);
-    const readResult = await conn.read(buf);
-    assert(readResult);
-    const msg = decoder.decode(buf.subarray(0, readResult));
-
-    assert(msg.includes("Content-Length: 0"));
-
-    conn.close();
-
-    ac.abort();
-    await server;
-  },
-);
-
 Deno.test(
   { permissions: { net: true } },
   async function httpServerConcurrentRequests() {

ext/http/01_http.js

@@ -31,8 +31,8 @@ import {
   _serverHandleIdleTimeout,
   WebSocket,
 } from "ext:deno_websocket/01_websocket.js";
-import { TcpConn, UnixConn } from "ext:deno_net/01_net.js";
-import { TlsConn } from "ext:deno_net/02_tls.js";
+import { listen, TcpConn, UnixConn } from "ext:deno_net/01_net.js";
+import { listenTls, TlsConn } from "ext:deno_net/02_tls.js";
 import {
   Deferred,
   getReadableStreamResourceBacking,
@@ -50,10 +50,13 @@ const {
   Set,
   SetPrototypeAdd,
   SetPrototypeDelete,
+  SetPrototypeClear,
   StringPrototypeCharCodeAt,
   StringPrototypeIncludes,
   StringPrototypeToLowerCase,
   StringPrototypeSplit,
+  SafeSet,
+  PromisePrototypeCatch,
   Symbol,
   SymbolAsyncIterator,
   TypeError,
@@ -554,4 +557,231 @@ function buildCaseInsensitiveCommaValueFinder(checkText) {
 internals.buildCaseInsensitiveCommaValueFinder =
   buildCaseInsensitiveCommaValueFinder;
 
-export { _ws, HttpConn, upgradeHttp, upgradeHttpRaw, upgradeWebSocket };
+function hostnameForDisplay(hostname) {
+  // If the hostname is "0.0.0.0", we display "localhost" in console
+  // because browsers in Windows don't resolve "0.0.0.0".
+  // See the discussion in https://github.com/denoland/deno_std/issues/1165
+  return hostname === "0.0.0.0" ? "localhost" : hostname;
+}
+
+async function respond(handler, requestEvent, connInfo, onError) {
+  let response;
+
+  try {
+    response = await handler(requestEvent.request, connInfo);
+
+    if (response.bodyUsed && response.body !== null) {
+      throw new TypeError("Response body already consumed.");
+    }
+  } catch (e) {
+    // Invoke `onError` handler if the request handler throws.
+    response = await onError(e);
+  }
+
+  try {
+    // Send the response.
+    await requestEvent.respondWith(response);
+  } catch {
+    // `respondWith()` can throw for various reasons, including downstream and
+    // upstream connection errors, as well as errors thrown during streaming
+    // of the response content.  In order to avoid false negatives, we ignore
+    // the error here and let `serveHttp` close the connection on the
+    // following iteration if it is in fact a downstream connection error.
+  }
+}
+
+async function serveConnection(
+  server,
+  activeHttpConnections,
+  handler,
+  httpConn,
+  connInfo,
+  onError,
+) {
+  while (!server.closed) {
+    let requestEvent = null;
+
+    try {
+      // Yield the new HTTP request on the connection.
+      requestEvent = await httpConn.nextRequest();
+    } catch {
+      // Connection has been closed.
+      break;
+    }
+
+    if (requestEvent === null) {
+      break;
+    }
+
+    respond(handler, requestEvent, connInfo, onError);
+  }
+
+  SetPrototypeDelete(activeHttpConnections, httpConn);
+  try {
+    httpConn.close();
+  } catch {
+    // Connection has already been closed.
+  }
+}
+
+async function serve(arg1, arg2) {
+  let options = undefined;
+  let handler = undefined;
+  if (typeof arg1 === "function") {
+    handler = arg1;
+    options = arg2;
+  } else if (typeof arg2 === "function") {
+    handler = arg2;
+    options = arg1;
+  } else {
+    options = arg1;
+  }
+  if (handler === undefined) {
+    if (options === undefined) {
+      throw new TypeError(
+        "No handler was provided, so an options bag is mandatory.",
+      );
+    }
+    handler = options.handler;
+  }
+  if (typeof handler !== "function") {
+    throw new TypeError("A handler function must be provided.");
+  }
+  if (options === undefined) {
+    options = {};
+  }
+
+  const signal = options.signal;
+  const onError = options.onError ?? function (error) {
+    console.error(error);
+    return new Response("Internal Server Error", { status: 500 });
+  };
+  const onListen = options.onListen ?? function ({ port }) {
+    console.log(
+      `Listening on http://${hostnameForDisplay(listenOpts.hostname)}:${port}/`,
+    );
+  };
+  const listenOpts = {
+    hostname: options.hostname ?? "127.0.0.1",
+    port: options.port ?? 9000,
+    reuseport: options.reusePort ?? false,
+  };
+
+  if (options.cert || options.key) {
+    if (!options.cert || !options.key) {
+      throw new TypeError(
+        "Both cert and key must be provided to enable HTTPS.",
+      );
+    }
+    listenOpts.cert = options.cert;
+    listenOpts.key = options.key;
+  }
+
+  let listener;
+  if (listenOpts.cert && listenOpts.key) {
+    listener = listenTls({
+      hostname: listenOpts.hostname,
+      port: listenOpts.port,
+      cert: listenOpts.cert,
+      key: listenOpts.key,
+    });
+  } else {
+    listener = listen({
+      hostname: listenOpts.hostname,
+      port: listenOpts.port,
+    });
+  }
+
+  const serverDeferred = new Deferred();
+  const activeHttpConnections = new SafeSet();
+
+  const server = {
+    transport: listenOpts.cert && listenOpts.key ? "https" : "http",
+    hostname: listenOpts.hostname,
+    port: listenOpts.port,
+    closed: false,
+
+    close() {
+      if (server.closed) {
+        return;
+      }
+      server.closed = true;
+      try {
+        listener.close();
+      } catch {
+        // Might have been already closed.
+      }
+
+      for (const httpConn of new SafeSetIterator(activeHttpConnections)) {
+        try {
+          httpConn.close();
+        } catch {
+          // Might have been already closed.
+        }
+      }
+
+      SetPrototypeClear(activeHttpConnections);
+      serverDeferred.resolve();
+    },
+
+    async serve() {
+      while (!server.closed) {
+        let conn;
+
+        try {
+          conn = await listener.accept();
+        } catch {
+          // Listener has been closed.
+          if (!server.closed) {
+            console.log("Listener has closed unexpectedly");
+          }
+          break;
+        }
+
+        let httpConn;
+        try {
+          const rid = ops.op_http_start(conn.rid);
+          httpConn = new HttpConn(rid, conn.remoteAddr, conn.localAddr);
+        } catch {
+          // Connection has been closed;
+          continue;
+        }
+
+        SetPrototypeAdd(activeHttpConnections, httpConn);
+
+        const connInfo = {
+          localAddr: conn.localAddr,
+          remoteAddr: conn.remoteAddr,
+        };
+        // Serve the HTTP connection
+        serveConnection(
+          server,
+          activeHttpConnections,
+          handler,
+          httpConn,
+          connInfo,
+          onError,
+        );
+      }
+      await serverDeferred.promise;
+    },
+  };
+
+  signal?.addEventListener(
+    "abort",
+    () => {
+      try {
+        server.close();
+      } catch {
+        // Pass
+      }
+    },
+    { once: true },
+  );
+
+  onListen(listener.addr);
+
+  await PromisePrototypeCatch(server.serve(), console.error);
+}
+
+export { _ws, HttpConn, serve, upgradeHttp, upgradeHttpRaw, upgradeWebSocket };

runtime/js/90_deno_ns.js

@@ -173,7 +173,7 @@ const denoNsUnstable = {
   funlockSync: fs.funlockSync,
   upgradeHttp: http.upgradeHttp,
   upgradeHttpRaw: flash.upgradeHttpRaw,
-  serve: flash.createServe(ops.op_flash_serve),
+  serve: http.serve,
   openKv: kv.openKv,
   Kv: kv.Kv,
   KvU64: kv.KvU64,