fix: complete HTTP incoming handler tests (#317)

packages/preview2-shim/lib/io/worker-http.js

@@ -39,6 +39,7 @@ export async function setOutgoingResponse(
       headers.map(([key, val]) => [key, textDecoder.decode(val)])
     )
   );
+  response.flushHeaders();
   const { stream } = getStreamOrThrow(streamId);
   stream.pipe(response);
   responses.delete(id);
@@ -55,8 +56,9 @@ export async function startHttpServer(id, { port, host }) {
       payload: {
         responseId,
         method: req.method,
+        host: req.headers.host || host || 'localhost',
         pathWithQuery: req.url,
-        headers: Object.entries(req.headers),
+        headers: Object.entries(req.headersDistinct).flatMap(([key, val]) => val.map(val => [key, val])),
         streamId,
       },
     });
@@ -71,7 +73,9 @@ export async function startHttpServer(id, { port, host }) {
 
 export async function createHttpRequest(
   method,
-  url,
+  scheme,
+  authority,
+  pathWithQuery,
   headers,
   bodyId,
   connectTimeout,
@@ -99,34 +103,35 @@ export async function createHttpRequest(
   }
   try {
     // Make a request
-    const parsedUrl = new URL(url);
     let req;
-    switch (parsedUrl.protocol) {
+    switch (scheme) {
       case "http:":
         req = httpRequest({
           agent: httpAgent,
           method,
-          host: parsedUrl.hostname,
-          port: parsedUrl.port,
-          path: parsedUrl.pathname + parsedUrl.search,
-          headers,
-          timeout: connectTimeout && Number(connectTimeout)
+          host: authority.split(':')[0],
+          port: authority.split(':')[1],
+          path: pathWithQuery,
+          timeout: connectTimeout && Number(connectTimeout),
         });
         break;
       case "https:":
         req = httpsRequest({
           agent: httpsAgent,
           method,
-          host: parsedUrl.hostname,
-          port: parsedUrl.port,
-          path: parsedUrl.pathname + parsedUrl.search,
-          headers,
-          timeout: connectTimeout && Number(connectTimeout)
+          host: authority.split(':')[0],
+          port: authority.split(':')[1],
+          path: pathWithQuery,
+          timeout: connectTimeout && Number(connectTimeout),
         });
         break;
       default:
         throw { tag: "HTTP-protocol-error" };
     }
+    for (const [key, value] of headers) {
+      req.appendHeader(key, value);
+    }
+    req.flushHeaders();
     if (stream) {
       stream.pipe(req);
     } else {

packages/preview2-shim/lib/io/worker-io.js

@@ -35,7 +35,6 @@ const workerPath = fileURLToPath(
   new URL("./worker-thread.js", import.meta.url)
 );
 
-
 const httpIncomingHandlers = new Map();
 export function registerIncomingHttpHandler (id, handler) {
   httpIncomingHandlers.set(id, handler);

packages/preview2-shim/lib/io/worker-thread.js

@@ -156,7 +156,9 @@ function handle(call, id, payload) {
     case HTTP_CREATE_REQUEST: {
       const {
         method,
-        url,
+        scheme,
+        authority,
+        pathWithQuery,
         headers,
         body,
         connectTimeout,
@@ -166,7 +168,9 @@ function handle(call, id, payload) {
       return createFuture(
         createHttpRequest(
           method,
-          url,
+          scheme,
+          authority,
+          pathWithQuery,
           headers,
           body,
           connectTimeout,

packages/preview2-shim/lib/nodejs/http.js

@@ -277,15 +277,17 @@ class OutgoingRequest {
     const betweenBytesTimeout = options?.betweenBytesTimeoutMs();
     const firstByteTimeout = options?.firstByteTimeoutMs();
     const scheme = schemeString(request.#scheme);
-    const url = scheme + request.#authority + (request.#pathWithQuery || "");
-    const headers = [["host", request.#authority]];
+    // note: host header is automatically added by Node.js
+    const headers = [];
     const decoder = new TextDecoder();
     for (const [key, value] of request.#headers.entries()) {
       headers.push([key, decoder.decode(value)]);
     }
     return futureIncomingResponseCreate(
       request.#method.val || request.#method.tag,
-      url,
+      scheme,
+      request.#authority,
+      request.#pathWithQuery,
       headers,
       outgoingBodyOutputStreamId(request.#body),
       connectTimeout,
@@ -417,11 +419,13 @@ class FutureIncomingResponse {
   [symbolDispose]() {
     if (this.#pollId) ioCall(FUTURE_DISPOSE | HTTP, this.#pollId);
   }
-  static _create(method, url, headers, body, connectTimeout, betweenBytesTimeout, firstByteTimeout) {
+  static _create(method, scheme, authority, pathWithQuery, headers, body, connectTimeout, betweenBytesTimeout, firstByteTimeout) {
     const res = new FutureIncomingResponse();
     res.#pollId = ioCall(HTTP_CREATE_REQUEST, null, {
       method,
-      url,
+      scheme,
+      authority,
+      pathWithQuery,
       headers,
       body,
       connectTimeout,
@@ -628,13 +632,13 @@ export class HTTPServer {
     }
     registerIncomingHttpHandler(
       this.#id,
-      ({ method, pathWithQuery, headers, responseId, streamId }) => {
+      ({ method, pathWithQuery, host, headers, responseId, streamId }) => {
         const textEncoder = new TextEncoder();
         const request = incomingRequestCreate(
           parseMethod(method),
           pathWithQuery,
           { tag: "HTTP" },
-          "authority.org",
+          host,
           fieldsLock(
             fieldsFromEntriesChecked(
               headers

tests/generated/proxy_hash.rs

@@ -2,16 +2,16 @@
 //! To regenerate this file re-run `cargo xtask generate tests` from the project root
 
 use std::fs;
-// use xshell::{cmd, Shell};
+use xshell::{cmd, Shell};
 
 #[test]
 fn proxy_hash() -> anyhow::Result<()> {
-    // let sh = Shell::new()?;
-    // let wasi_file = "./tests/rundir/proxy_hash.component.wasm";
+    let sh = Shell::new()?;
+    let wasi_file = "./tests/rundir/proxy_hash.component.wasm";
     let _ = fs::remove_dir_all("./tests/rundir/proxy_hash");
 
-    // let cmd = cmd!(sh, "node ./src/jco.js run  --jco-dir ./tests/rundir/proxy_hash --jco-import ./tests/virtualenvs/server-api-proxy-streaming.js {wasi_file} hello this '' 'is an argument' 'with 🚩 emoji'");
+    let cmd = cmd!(sh, "node ./src/jco.js run  --jco-dir ./tests/rundir/proxy_hash --jco-import ./tests/virtualenvs/server-api-proxy-streaming.js {wasi_file} hello this '' 'is an argument' 'with 🚩 emoji'");
 
-    // cmd.run()?;
-    panic!("skipped"); // Ok(())
+    cmd.run()?;
+    Ok(())
 }

tests/server/index.js

@@ -12,23 +12,35 @@ export const testDir = await mkdtemp(tmpdir());
 await $init;
 
 process.on("exit", () => {
-  // send stop message to server process
-  serverProcess.send(null);
+  // send stop message to server processes
+  for (const server of servers) {
+    server.send(null);
+  }
   rmdirSync(testDir, { recursive: true });
 });
 
-const serverProcess = fork(
-  fileURLToPath(import.meta.url.split("/").slice(0, -1).join("/")) +
-    "/http-server.js"
-);
-serverProcess.on("error", (err) => {
-  console.error("server process error", err);
-});
-const runningPromise = new Promise((resolve) =>
-  serverProcess.on("message", resolve)
+symlinkSync(
+  fileURLToPath(
+    import.meta.url.split("/").slice(0, -3).join("/") + "/node_modules"
+  ),
+  testDir + "/node_modules"
 );
+writeFileSync(testDir + "/package.json", '{"type":"module"}');
+
+const servers = [];
 
 export async function createIncomingServer(serverName) {
+  const serverProcess = fork(
+    fileURLToPath(import.meta.url.split("/").slice(0, -1).join("/")) +
+      "/http-server.js"
+  );
+  servers.push(serverProcess);
+  serverProcess.on("error", (err) => {
+    console.error("server process error", err);
+  });
+  const runningPromise = new Promise((resolve) =>
+    serverProcess.on("message", resolve)
+  );
   const componentPath = fileURLToPath(import.meta.url.split("/").slice(0, -2).join("/")) + `/rundir/${serverName}.component.wasm`;
   console.error("loading component " + componentPath);
   try {
@@ -46,13 +58,6 @@ export async function createIncomingServer(serverName) {
         "wasi:sockets/*": "@bytecodealliance/preview2-shim/sockets#*",
       }),
     });
-    symlinkSync(
-      fileURLToPath(
-        import.meta.url.split("/").slice(0, -3).join("/") + "/node_modules"
-      ),
-      testDir + "/node_modules"
-    );
-    writeFileSync(testDir + "/package.json", '{"type":"module"}');
     for (const [name, contents] of files) {
       writeFileSync(testDir + "/" + name, contents);
     }

tests/virtualenvs/server-api-proxy-streaming.js

@@ -1,8 +1,10 @@
 import { _appendEnv } from "@bytecodealliance/preview2-shim/cli";
 import { createIncomingServer } from "../server/index.js";
 
-const authority = await createIncomingServer('api_proxy_streaming');
+const authority1 = await createIncomingServer('api_proxy_streaming');
+const authority2 = await createIncomingServer('api_proxy_streaming');
 
 _appendEnv({
-  "HANDLER_API_PROXY_STREAMING": authority
+  "HANDLER_API_PROXY_STREAMING1": authority1,
+  "HANDLER_API_PROXY_STREAMING2": authority2
 });

xtask/src/generate/tests.rs

@@ -3,14 +3,13 @@ use xshell::{cmd, Shell};
 
 // for debugging
 const TRACE: bool = false;
-const TEST_FILTER: &[&str] = &[];
-/* &[
-    "proxy_handler",
-    "proxy_echo",
-    "proxy_hash",
-    "api_proxy",
-    "api_proxy_streaming",
-]; */
+const TEST_FILTER: &[&str] = &[]; /*&[
+                                      "proxy_handler",
+                                      "proxy_echo",
+                                      "proxy_hash",
+                                      "api_proxy",
+                                      "api_proxy_streaming",
+                                  ];*/
 
 pub fn run() -> anyhow::Result<()> {
     let sh = Shell::new()?;
@@ -112,7 +111,7 @@ fn generate_test(test_name: &str) -> String {
 
     let skip = match test_name {
         // these tests currently stall
-        "api_read_only" | "preview1_path_open_read_write" | "proxy_hash" => true,
+        "api_read_only" | "preview1_path_open_read_write" => true,
         _ => false,
     };