update error handling in stream2, add abort controller

Author: vhess

lib/http-proxy/passes/web-incoming.ts

@@ -69,7 +69,7 @@ export function XHeaders(req: Request, _res: Response, options: ServerOptions) {
       (req.headers["x-forwarded-" + header] || "") + (req.headers["x-forwarded-" + header] ? "," : "") + values[header];
   }
 
-  req.headers["x-forwarded-host"] = req.headers["x-forwarded-host"] || req.headers["host"] || "";
+  req.headers["x-forwarded-host"] = req.headers["x-forwarded-host"] || req.headers["host"] || req.headers[":authority"] || "";
 }
 
 // Does the actual proxying. If `forward` is enabled fires up
@@ -208,6 +208,12 @@ async function stream2(
 ) {
   // Helper function to handle errors consistently throughout the fetch path
   const handleError = (err: Error, target?: ProxyTargetUrl) => {
+    const e = err as any;
+    // Copy code from cause if available and missing on err
+    if (e.code === undefined && e.cause?.code) {
+      e.code = e.cause.code;
+    }
+
     if (cb) {
       cb(err, req, res, target);
     } else {
@@ -229,9 +235,27 @@ async function stream2(
   const customFetch = options.fetch || fetch;
   const fetchOptions = options.fetchOptions ?? {} as FetchOptions;
 
+  const controller = new AbortController();
+  const { signal } = controller;
+
+  if (options.proxyTimeout) {
+    setTimeout(() => {
+      controller.abort();
+    }, options.proxyTimeout);
+  }
+
+  // Ensure we abort proxy if request is aborted
+  res.on("close", () => {
+    const aborted = !res.writableFinished;
+    if (aborted) {
+      controller.abort();
+    }
+  });
+
   const prepareRequest = (outgoing: common.Outgoing) => {
     const requestOptions: RequestInit = {
       method: outgoing.method,
+      signal,
       ...fetchOptions.requestOptions,
     };
 
@@ -294,6 +318,16 @@ async function stream2(
         }
       }
     } catch (err) {
+      if ((err as Error).name === "AbortError") {
+        // Handle aborts (timeout or client disconnect)
+        if (options.proxyTimeout && signal.aborted) {
+          const proxyTimeoutErr = new Error("Proxy timeout");
+          (proxyTimeoutErr as any).code = "ECONNRESET";
+          handleError(proxyTimeoutErr, options.forward);
+        }
+        // If aborted by client (res.close), we might not want to emit an error or maybe just log it
+        return;
+      }
       handleError(err as Error, options.forward);
     }
 
@@ -385,6 +419,14 @@ async function stream2(
       server?.emit("end", req, res, fakeProxyRes);
     }
   } catch (err) {
+    if ((err as Error).name === "AbortError") {
+      if (options.proxyTimeout && signal.aborted) {
+        const proxyTimeoutErr = new Error("Proxy timeout");
+        (proxyTimeoutErr as any).code = "ECONNRESET";
+        handleError(proxyTimeoutErr, options.target);
+      }
+      return;
+    }
     handleError(err as Error, options.target);
   }
 }

lib/test/http/fetch-timeout.test.ts

@@ -0,0 +1,62 @@
+
+import * as http from "node:http";
+import * as httpProxy from "../..";
+import getPort from "../get-port";
+import { describe, it, expect, beforeAll, afterAll } from "vitest";
+import { fetch } from "undici";
+
+describe("Fetch Proxy Timeout", () => {
+  let ports: Record<"http" | "proxy", number>;
+  beforeAll(async () => {
+    ports = { http: await getPort(), proxy: await getPort() };
+  });
+
+  const servers: Record<string, any> = {};
+
+  it("Create the target HTTP server that hangs", async () => {
+    servers.http = http
+      .createServer((_req, _res) => {
+        // Do nothing, let it hang
+      })
+      .listen(ports.http);
+  });
+
+  it("Create the proxy server with fetch and timeout", async () => {
+    servers.proxy = httpProxy
+      .createServer({
+        target: `http://localhost:${ports.http}`,
+        fetch: fetch, // Enable fetch path
+        proxyTimeout: 500, // 500ms timeout
+      })
+      .listen(ports.proxy);
+  });
+
+  it("should timeout the request and emit error", async () => {
+    return new Promise<void>((resolve, reject) => {
+      const timeout = setTimeout(() => {
+        reject(new Error("Test timed out"));
+      }, 2000);
+
+      servers.proxy.once('error', (err: Error, _req: any, res: any) => {
+        clearTimeout(timeout);
+        try {
+          expect(err).toBeTruthy();
+          expect(err.message).toBe("Proxy timeout");
+          res.statusCode = 504;
+          res.end("Gateway Timeout");
+          resolve();
+        } catch (e) {
+          reject(e);
+        }
+      });
+
+      fetch(`http://localhost:${ports.proxy}`).catch(() => {
+         // Ignore client side fetch error, we care about server side error emission
+      });
+    });
+  });
+
+  afterAll(async () => {
+    Object.values(servers).map((x: any) => x?.close());
+  });
+});

lib/test/lib/http-proxy-passes-web-incoming.test.ts

@@ -119,7 +119,11 @@ describe("#createProxyServer.web() using own http server", () => {
       const source = http.createServer((req, res) => {
         res.end();
         expect(req.method).toEqual("GET");
-        expect(req.headers.host?.split(":")[1]).toEqual(`${ports["8081"]}`);
+        if (process.env.FORCE_FETCH_PATH === "true") {
+          expect(req.headers.host?.split(":")[1]).toEqual(`${ports["8080"]}`);
+        } else {
+          expect(req.headers.host?.split(":")[1]).toEqual(`${ports["8081"]}`);
+        }
       });
 
       proxyServer.listen(ports["8081"]);