fix(ext/node): work correctly with wrapper Response objects, use correct rawHeaders structure (#29056)

Fixes #28022

Basically drizzle-kit studio uses hono with the node-server adapter.
That creates wrapper objects for responses that forward property getters
to the underlying response (the one we provided). However, in deno.serve
we were assuming that the response was actually the same response we
initially gave and crashed when it wasn't. instead, just call the
property getters if we can't find the inner response.

The raw headers bug is that we were exposing the `rawHeaders` field on
`Incoming` as a `Headers` object, instead it's supposed to be a flat
array of the header keys + values. I.e. `["Content-Type:",
"application/json", "Host:", "http://localhost"]`

Author: nathanwhit

ext/http/00_serve.ts

@@ -58,6 +58,7 @@ import {
   ResponsePrototype,
   toInnerResponse,
 } from "ext:deno_fetch/23_response.js";
+import { headerListFromHeaders } from "ext:deno_fetch/20_headers.js";
 import {
   abortRequest,
   fromInnerRequest,
@@ -500,8 +501,15 @@ function fastSyncResponseOrStream(
     return;
   }
 
-  const stream = respBody.streamOrStatic;
-  const body = stream.body;
+  let stream;
+  let body;
+  if (respBody.streamOrStatic) {
+    stream = respBody.streamOrStatic;
+    body = stream.body;
+  } else {
+    stream = respBody;
+    body = respBody;
+  }
   if (body !== undefined) {
     // We ensure the response has not been consumed yet in the caller of this
     // function.
@@ -632,8 +640,19 @@ function mapToCallback(context, callback, onError) {
       return;
     }
 
-    const status = inner.status;
-    const headers = inner.headerList;
+    let status;
+    let headers;
+    let body;
+    if (inner) {
+      status = inner.status;
+      headers = inner.headerList;
+      body = inner.body;
+    } else {
+      status = response.status;
+      headers = headerListFromHeaders(response.headers);
+      body = response.body;
+    }
+
     if (headers && headers.length > 0) {
       if (headers.length == 1) {
         op_http_set_response_header(req, headers[0][0], headers[0][1]);
@@ -642,7 +661,7 @@ function mapToCallback(context, callback, onError) {
       }
     }
 
-    fastSyncResponseOrStream(req, inner.body, status, innerRequest);
+    fastSyncResponseOrStream(req, body, status, innerRequest);
   };
 
   if (TRACING_ENABLED) {

ext/node/polyfills/http.ts

@@ -1780,6 +1780,8 @@ Object.defineProperty(ServerResponse.prototype, "connection", {
   ),
 });
 
+const kRawHeaders = Symbol("rawHeaders");
+
 // TODO(@AaronO): optimize
 export class IncomingMessageForServer extends NodeReadable {
   #headers: Record<string, string>;
@@ -1819,7 +1821,7 @@ export class IncomingMessageForServer extends NodeReadable {
     this.method = "";
     this.socket = socket;
     this.upgrade = null;
-    this.rawHeaders = [];
+    this[kRawHeaders] = [];
     socket?.on("error", (e) => {
       if (this.listenerCount("error") > 0) {
         this.emit("error", e);
@@ -1842,7 +1844,7 @@ export class IncomingMessageForServer extends NodeReadable {
   get headers() {
     if (!this.#headers) {
       this.#headers = {};
-      const entries = headersEntries(this.rawHeaders);
+      const entries = headersEntries(this[kRawHeaders]);
       for (let i = 0; i < entries.length; i++) {
         const entry = entries[i];
         this.#headers[entry[0]] = entry[1];
@@ -1855,6 +1857,16 @@ export class IncomingMessageForServer extends NodeReadable {
     this.#headers = val;
   }
 
+  get rawHeaders() {
+    const entries = headersEntries(this[kRawHeaders]);
+    const out = new Array(entries.length * 2);
+    for (let i = 0; i < entries.length; i++) {
+      out[i * 2] = entries[i][0];
+      out[i * 2 + 1] = entries[i][1];
+    }
+    return out;
+  }
+
   // connection is deprecated, but still tested in unit test.
   get connection() {
     return this.socket;
@@ -1959,7 +1971,7 @@ export class ServerImpl extends EventEmitter {
       req.upgrade =
         request.headers.get("connection")?.toLowerCase().includes("upgrade") &&
         request.headers.get("upgrade");
-      req.rawHeaders = request.headers;
+      req[kRawHeaders] = request.headers;
 
       if (req.upgrade && this.listenerCount("upgrade") > 0) {
         const { conn, response } = upgradeHttpRaw(request);

tests/specs/node/wrapped_http_response/__test__.jsonc

@@ -0,0 +1,8 @@
+{
+  "tempDir": true,
+
+  "steps": [{
+    "args": "run -A main.ts",
+    "output": "done\n"
+  }]
+}

tests/specs/node/wrapped_http_response/main.ts

@@ -0,0 +1,54 @@
+// Adapted from https://github.com/honojs/node-server/blob/1eb73c6d985665e75458ddd08c23bbc1dbdc7bcd/src/listener.ts
+// and https://github.com/honojs/node-server/blob/1eb73c6d985665e75458ddd08c23bbc1dbdc7bcd/src/server.ts
+import {
+  buildOutgoingHttpHeaders,
+  Response as WrappedResponse,
+} from "./response.ts";
+import { createServer, OutgoingHttpHeaders, ServerResponse } from "node:http";
+
+Object.defineProperty(globalThis, "Response", {
+  value: WrappedResponse,
+});
+
+const { promise, resolve } = Promise.withResolvers<void>();
+
+const responseViaResponseObject = async (
+  res: Response,
+  outgoing: ServerResponse,
+) => {
+  const resHeaderRecord: OutgoingHttpHeaders = buildOutgoingHttpHeaders(
+    res.headers,
+  );
+
+  if (res.body) {
+    const buffer = await res.arrayBuffer();
+    resHeaderRecord["content-length"] = buffer.byteLength;
+
+    outgoing.writeHead(res.status, resHeaderRecord);
+    outgoing.end(new Uint8Array(buffer));
+  } else {
+    outgoing.writeHead(res.status, resHeaderRecord);
+    outgoing.end();
+  }
+};
+
+const server = createServer((_req, res) => {
+  const response = new Response("Hello, world!");
+  return responseViaResponseObject(response, res);
+});
+
+using _server = {
+  [Symbol.dispose]() {
+    server.close();
+  },
+};
+
+server.listen(0, async () => {
+  const { port } = server.address() as { port: number };
+  const response = await fetch(`http://localhost:${port}`);
+  await response.text();
+  resolve();
+});
+
+await promise;
+console.log("done");

tests/specs/node/wrapped_http_response/response.ts

@@ -0,0 +1,113 @@
+// Adapted from https://github.com/honojs/node-server/blob/1eb73c6d985665e75458ddd08c23bbc1dbdc7bcd/src/response.ts
+// deno-lint-ignore-file no-explicit-any
+//
+import type { OutgoingHttpHeaders } from "node:http";
+
+interface InternalBody {
+  source: string | Uint8Array | FormData | Blob | null;
+  stream: ReadableStream;
+  length: number | null;
+}
+
+const GlobalResponse = globalThis.Response;
+
+const responseCache = Symbol("responseCache");
+const getResponseCache = Symbol("getResponseCache");
+export const cacheKey = Symbol("cache");
+
+export const buildOutgoingHttpHeaders = (
+  headers: Headers | HeadersInit | null | undefined,
+): OutgoingHttpHeaders => {
+  const res: OutgoingHttpHeaders = {};
+  if (!(headers instanceof Headers)) {
+    headers = new Headers(headers ?? undefined);
+  }
+
+  const cookies = [];
+  for (const [k, v] of headers) {
+    if (k === "set-cookie") {
+      cookies.push(v);
+    } else {
+      res[k] = v;
+    }
+  }
+  if (cookies.length > 0) {
+    res["set-cookie"] = cookies;
+  }
+  res["content-type"] ??= "text/plain; charset=UTF-8";
+
+  return res;
+};
+
+export class Response {
+  #body?: BodyInit | null;
+  #init?: ResponseInit;
+
+  [getResponseCache](): typeof GlobalResponse {
+    delete (this as any)[cacheKey];
+    return ((this as any)[responseCache] ||= new GlobalResponse(
+      this.#body,
+      this.#init,
+    ));
+  }
+
+  constructor(body?: BodyInit | null, init?: ResponseInit) {
+    this.#body = body;
+    if (init instanceof Response) {
+      const cachedGlobalResponse = (init as any)[responseCache];
+      if (cachedGlobalResponse) {
+        this.#init = cachedGlobalResponse;
+        // instantiate GlobalResponse cache and this object always returns value from global.Response
+        this[getResponseCache]();
+        return;
+      } else {
+        this.#init = init.#init;
+      }
+    } else {
+      this.#init = init;
+    }
+
+    if (
+      typeof body === "string" ||
+      typeof (body as ReadableStream)?.getReader !== "undefined"
+    ) {
+      let headers =
+        (init?.headers || { "content-type": "text/plain; charset=UTF-8" }) as
+          | Record<string, string>
+          | Headers
+          | OutgoingHttpHeaders;
+      if (headers instanceof Headers) {
+        headers = buildOutgoingHttpHeaders(headers);
+      }
+
+      (this as any)[cacheKey] = [init?.status || 200, body, headers];
+    }
+  }
+}
+[
+  "body",
+  "bodyUsed",
+  "headers",
+  "ok",
+  "redirected",
+  "status",
+  "statusText",
+  "trailers",
+  "type",
+  "url",
+].forEach((k) => {
+  Object.defineProperty(Response.prototype, k, {
+    get() {
+      return this[getResponseCache]()[k];
+    },
+  });
+});
+["arrayBuffer", "blob", "clone", "formData", "json", "text"].forEach((k) => {
+  Object.defineProperty(Response.prototype, k, {
+    value: function () {
+      return this[getResponseCache]()[k]();
+    },
+  });
+});
+Object.setPrototypeOf(Response, GlobalResponse);
+Object.setPrototypeOf(Response.prototype, GlobalResponse.prototype);

tests/unit_node/http_test.ts

@@ -2041,3 +2041,42 @@ Deno.test("[node/http] 'close' event is emitted on ServerResponse object when th
   await new Promise((resolve) => server.close(resolve));
   assert(responseCloseEmitted);
 });
+
+Deno.test("[node/http] rawHeaders are in flattened format", async () => {
+  const getHeader = (req: IncomingMessage, name: string) => {
+    const idx = req.rawHeaders.indexOf(name);
+    if (idx < 0) {
+      throw new Error(`Header ${name} not found`);
+    }
+    return [name, req.rawHeaders[idx + 1]];
+  };
+  const { promise, resolve } = Promise.withResolvers<void>();
+  const server = http.createServer((req, res) => {
+    resolve();
+    // TODO(nathanwhit): the raw headers should not be lowercased, they should be
+    // exactly as they appeared in the request
+    assertEquals(getHeader(req, "content-type"), [
+      "content-type",
+      "text/plain",
+    ]);
+    assertEquals(getHeader(req, "set-cookie"), [
+      "set-cookie",
+      "foo=bar",
+    ]);
+    res.end();
+  });
+
+  server.listen(0, async () => {
+    const { port } = server.address() as { port: number };
+    const response = await fetch(`http://localhost:${port}`, {
+      headers: {
+        "Set-Cookie": "foo=bar",
+        "Content-Type": "text/plain",
+      },
+    });
+    await response.body?.cancel();
+  });
+
+  await promise;
+  await new Promise((resolve) => server.close(resolve));
+});