Fix signal abort in H/1 handler

Signed-off-by: Sri Krishna Paritala <skrishna@buf.build>
connectrpc · Sep 9, 2024 · 99de3f4 · 99de3f4
1 parent 90b2bb9
commit 99de3f4
Show file tree

Hide file tree

Showing 3 changed files with 111 additions and 62 deletions.
diff --git a/packages/connect-node/src/connect-node-adapter.ts b/packages/connect-node/src/connect-node-adapter.ts
@@ -21,7 +21,7 @@ import type {
 import type { UniversalHandler } from "@connectrpc/connect/protocol";
 import { uResponseNotFound } from "@connectrpc/connect/protocol";
 import {
-  universalRequestFromNodeRequest,
+  universalRequestFromNodeResponse,
   universalResponseToNodeResponse,
 } from "./node-universal-handler.js";
 import type {
@@ -94,8 +94,8 @@ export function connectNodeAdapter(
       (options.fallback ?? fallback)(req, res);
       return;
     }
-    const uReq = universalRequestFromNodeRequest(
-      req,
+    const uReq = universalRequestFromNodeResponse(
+      res,
       undefined,
       options.contextValues?.(req),
     );

diff --git a/packages/connect-node/src/node-universal-handler.spec.ts b/packages/connect-node/src/node-universal-handler.spec.ts
@@ -15,7 +15,10 @@
 import { useNodeServer } from "./use-node-server-helper.spec.js";
 import * as http2 from "http2";
 import * as http from "http";
-import { universalRequestFromNodeRequest } from "./node-universal-handler.js";
+import {
+  universalRequestFromNodeRequest,
+  universalRequestFromNodeResponse,
+} from "./node-universal-handler.js";
 import { ConnectError } from "@connectrpc/connect";
 import { getNodeErrorProps } from "./node-error.js";
 import {
@@ -27,7 +30,7 @@ import type { UniversalServerRequest } from "@connectrpc/connect/protocol";
 // Polyfill the Headers API for Node versions < 18
 import "./node-headers-polyfill.js";
 
-describe("universalRequestFromNodeRequest()", function () {
+describe("universalRequestFromNodeResponse()", function () {
   describe("with HTTP/2 stream closed with an RST code", function () {
     let serverRequest: UniversalServerRequest | undefined;
     const server = useNodeServer(() => {
@@ -328,9 +331,9 @@ describe("universalRequestFromNodeRequest()", function () {
       | http.ServerResponse<http.IncomingMessage>
       | undefined;
     const server = useNodeServer(() =>
-      http.createServer(function (request, response) {
-        serverRequest = universalRequestFromNodeRequest(
-          request,
+      http.createServer(function (_, response) {
+        serverRequest = universalRequestFromNodeResponse(
+          response,
           undefined,
           undefined,
         );

diff --git a/packages/connect-node/src/node-universal-handler.ts b/packages/connect-node/src/node-universal-handler.ts
@@ -66,69 +66,41 @@ export type NodeServerResponse = (
 };
 
 /**
- * Converts a UniversalServerRequest to a Node.js server request.
+ * Converts a Node.js server response to a UniversalServerRequest.
  * This function helps to implement adapters to server frameworks running
  * on Node.js. Please be careful using this function in your own code, as we
  * may have to make changes to it in the future.
  */
+export function universalRequestFromNodeResponse(
+  nodeResponse: NodeServerResponse,
+  parsedJsonBody: JsonValue | undefined,
+  contextValues: ContextValues | undefined,
+): UniversalServerRequest {
+  return universalRequestFromNodeRequestOrResponse(
+    nodeResponse,
+    parsedJsonBody,
+    contextValues,
+  );
+}
+
+/**
+ * Converts a Node.js server request to a UniversalServerRequest.
+ * This function helps to implement adapters to server frameworks running
+ * on Node.js. Please be careful using this function in your own code, as we
+ * may have to make changes to it in the future.
+ *
+ * @deprecated Please use `universalRequestFromNodeResponse`
+ */
 export function universalRequestFromNodeRequest(
   nodeRequest: NodeServerRequest,
   parsedJsonBody: JsonValue | undefined,
   contextValues: ContextValues | undefined,
 ): UniversalServerRequest {
-  const encrypted =
-    "encrypted" in nodeRequest.socket && nodeRequest.socket.encrypted;
-  const protocol = encrypted ? "https" : "http";
-  const authority =
-    "authority" in nodeRequest
-      ? nodeRequest.authority
-      : nodeRequest.headers.host;
-  const pathname = nodeRequest.url ?? "";
-  if (authority === undefined) {
-    throw new ConnectError(
-      "unable to determine request authority from Node.js server request",
-      Code.Internal,
-    );
-  }
-  const body =
-    parsedJsonBody !== undefined
-      ? parsedJsonBody
-      : asyncIterableFromNodeServerRequest(nodeRequest);
-  const abortController = new AbortController();
-  if ("stream" in nodeRequest) {
-    // HTTP/2 has error codes we want to honor
-    nodeRequest.once("close", () => {
-      const err = connectErrorFromH2ResetCode(nodeRequest.stream.rstCode);
-      if (err !== undefined) {
-        abortController.abort(err);
-      } else {
-        abortController.abort();
-      }
-    });
-  } else {
-    // HTTP/1.1 does not have error codes, but Node.js has ECONNRESET
-    const onH1Error = (e: Error) => {
-      nodeRequest.off("error", onH1Error);
-      nodeRequest.off("close", onH1Close);
-      abortController.abort(connectErrorFromNodeReason(e));
-    };
-    const onH1Close = () => {
-      nodeRequest.off("error", onH1Error);
-      nodeRequest.off("close", onH1Close);
-      abortController.abort();
-    };
-    nodeRequest.once("error", onH1Error);
-    nodeRequest.once("close", onH1Close);
-  }
-  return {
-    httpVersion: nodeRequest.httpVersion,
-    method: nodeRequest.method ?? "",
-    url: new URL(pathname, `${protocol}://${authority}`).toString(),
-    header: nodeHeaderToWebHeader(nodeRequest.headers),
-    body,
-    signal: abortController.signal,
-    contextValues: contextValues,
-  };
+  return universalRequestFromNodeRequestOrResponse(
+    nodeRequest,
+    parsedJsonBody,
+    contextValues,
+  );
 }
 
 /**
@@ -204,6 +176,80 @@ export async function universalResponseToNodeResponse(
   }
 }
 
+function universalRequestFromNodeRequestOrResponse(
+  nodeRequestOrResponse: NodeServerResponse | NodeServerRequest,
+  parsedJsonBody: JsonValue | undefined,
+  contextValues: ContextValues | undefined,
+): UniversalServerRequest {
+  let nodeRequest: NodeServerRequest;
+  let nodeResponse: NodeServerResponse | undefined;
+  if ("req" in nodeRequestOrResponse) {
+    nodeRequest = nodeRequestOrResponse.req;
+    nodeResponse = nodeRequestOrResponse;
+  } else {
+    nodeRequest = nodeRequestOrResponse;
+  }
+  const encrypted =
+    "encrypted" in nodeRequest.socket && nodeRequest.socket.encrypted;
+  const protocol = encrypted ? "https" : "http";
+  const authority =
+    "authority" in nodeRequest
+      ? nodeRequest.authority
+      : nodeRequest.headers.host;
+  const pathname = nodeRequest.url ?? "";
+  if (authority === undefined) {
+    throw new ConnectError(
+      "unable to determine request authority from Node.js server request",
+      Code.Internal,
+    );
+  }
+  const body =
+    parsedJsonBody !== undefined
+      ? parsedJsonBody
+      : asyncIterableFromNodeServerRequest(nodeRequest);
+  const abortController = new AbortController();
+  if ("stream" in nodeRequest) {
+    // HTTP/2 has error codes we want to honor
+    nodeRequest.once("close", () => {
+      const err = connectErrorFromH2ResetCode(nodeRequest.stream.rstCode);
+      if (err !== undefined) {
+        abortController.abort(err);
+      } else {
+        abortController.abort();
+      }
+    });
+  } else {
+    // HTTP/1.1 does not have error codes, but Node.js has ECONNRESET
+    let closeHandler: http.IncomingMessage | http.ServerResponse;
+    if (nodeResponse === undefined) {
+      closeHandler = nodeRequest;
+    } else {
+      closeHandler = nodeResponse as http.ServerResponse;
+    }
+    const onH1Error = (e: Error) => {
+      nodeRequest.off("error", onH1Error);
+      closeHandler.off("close", onH1Close);
+      abortController.abort(connectErrorFromNodeReason(e));
+    };
+    const onH1Close = () => {
+      nodeRequest.off("error", onH1Error);
+      closeHandler.off("close", onH1Close);
+      abortController.abort();
+    };
+    nodeRequest.once("error", onH1Error);
+    closeHandler.once("close", onH1Close);
+  }
+  return {
+    httpVersion: nodeRequest.httpVersion,
+    method: nodeRequest.method ?? "",
+    url: new URL(pathname, `${protocol}://${authority}`).toString(),
+    header: nodeHeaderToWebHeader(nodeRequest.headers),
+    body,
+    signal: abortController.signal,
+    contextValues: contextValues,
+  };
+}
+
 async function* asyncIterableFromNodeServerRequest(
   request: NodeServerRequest,
 ): AsyncIterable<Uint8Array> {