Update HandlerContext signal to trigger on other reasons than timeouts

packages/connect-node-test/src/helpers/test-routes.ts

@@ -69,7 +69,7 @@ const testService: ServiceImpl<typeof TestService> = {
     );
     for (const param of request.responseParameters) {
       await maybeDelayResponse(param);
-      context.deadline?.throwIfAborted();
+      context.signal.throwIfAborted();
       yield {
         payload: interop.makeServerPayload(request.responseType, param.size),
       };
@@ -85,7 +85,7 @@ const testService: ServiceImpl<typeof TestService> = {
     );
     for (const param of request.responseParameters) {
       await maybeDelayResponse(param);
-      context.deadline?.throwIfAborted();
+      context.signal.throwIfAborted();
       yield {
         payload: interop.makeServerPayload(request.responseType, param.size),
       };
@@ -119,7 +119,7 @@ const testService: ServiceImpl<typeof TestService> = {
     for await (const req of requests) {
       for (const param of req.responseParameters) {
         await maybeDelayResponse(param);
-        context.deadline?.throwIfAborted();
+        context.signal.throwIfAborted();
         yield {
           payload: interop.makeServerPayload(req.responseType, param.size),
         };
@@ -141,7 +141,7 @@ const testService: ServiceImpl<typeof TestService> = {
     for await (const req of buffer) {
       for (const param of req.responseParameters) {
         await maybeDelayResponse(param);
-        context.deadline?.throwIfAborted();
+        context.signal.throwIfAborted();
         yield {
           payload: interop.makeServerPayload(req.responseType, param.size),
         };

packages/connect-node/src/node-universal-handler.ts

@@ -25,7 +25,10 @@ import {
   nodeHeaderToWebHeader,
   webHeaderToNodeHeaders,
 } from "./node-universal-header.js";
-import { connectErrorFromNodeReason } from "./node-error.js";
+import {
+  connectErrorFromH2ResetCode,
+  connectErrorFromNodeReason,
+} from "./node-error.js";
 
 /**
  * NodeHandlerFn is compatible with http.RequestListener and its equivalent
@@ -89,12 +92,28 @@ export function universalRequestFromNodeRequest(
     parsedJsonBody !== undefined
       ? parsedJsonBody
       : asyncIterableFromNodeServerRequest(nodeRequest);
+  const abortController = new AbortController();
+  if ("stream" in nodeRequest) {
+    // HTTP/2 has error codes we want to honor
+    nodeRequest.on("close", () => {
+      const err = connectErrorFromH2ResetCode(nodeRequest.stream.rstCode);
+      if (err !== undefined) {
+        abortController.abort(err);
+      } else {
+        abortController.abort();
+      }
+    });
+  } else {
+    // HTTP/1.1 does not
+    nodeRequest.on("close", () => abortController.abort());
+  }
   return {
     httpVersion: nodeRequest.httpVersion,
     method: nodeRequest.method ?? "",
     url: new URL(pathname, `${protocol}://${authority}`),
     header: nodeHeaderToWebHeader(nodeRequest.headers),
     body,
+    signal: abortController.signal,
   };
 }
 

packages/connect/src/implementation.ts

@@ -24,6 +24,7 @@ import type {
 } from "@bufbuild/protobuf";
 import { ConnectError } from "./connect-error.js";
 import { Code } from "./code.js";
+import { createLinkedAbortController } from "./protocol/linked-abort-controller.js";
 
 // prettier-ignore
 /**
@@ -71,8 +72,10 @@ export interface HandlerContext {
    */
   readonly service: ServiceType;
 
-  // TODO
-  readonly deadline?: AbortSignal;
+  /**
+   * An AbortSignal to know when the connection with the client is closed.
+   */
+  readonly signal: AbortSignal;
 
   /**
    * Incoming request headers.
@@ -99,14 +102,17 @@ export interface HandlerContext {
 export function createHandlerContext(
   spec: { service: ServiceType; method: MethodInfo },
   deadline: AbortSignal | undefined, // TODO
+  signal: AbortSignal,
   requestHeader: HeadersInit,
   responseHeader: HeadersInit,
   responseTrailer: HeadersInit
 ): HandlerContext {
+  const linkedAbortcontroller = createLinkedAbortController(deadline, signal);
+
   return {
     method: spec.method,
     service: spec.service,
-    deadline,
+    signal: linkedAbortcontroller.signal,
     requestHeader: new Headers(requestHeader),
     responseHeader: new Headers(responseHeader),
     responseTrailer: new Headers(responseTrailer),

packages/connect/src/protocol-connect/handler-factory.spec.ts

@@ -147,6 +147,7 @@ describe("createHandlerFactory()", function () {
           url: new URL("https://example.com"),
           header: new Headers({ "Content-Type": "application/json" }),
           body: 777,
+          signal: new AbortController().signal,
         });
         expect(res.status).toBe(400);
         expect(res.body).toBeInstanceOf(Uint8Array);
@@ -171,6 +172,7 @@ describe("createHandlerFactory()", function () {
             "Connect-Protocol-Version": "UNEXPECTED",
           }),
           body: 777,
+          signal: new AbortController().signal,
         });
         expect(res.status).toBe(400);
         expect(res.body).toBeInstanceOf(Uint8Array);
@@ -202,6 +204,7 @@ describe("createHandlerFactory()", function () {
           url: new URL("https://example.com"),
           header: new Headers({ "Content-Type": "application/connect+json" }),
           body: createAsyncIterableBytes(new Uint8Array()),
+          signal: new AbortController().signal,
         });
         expect(res.status).toBe(200);
         expect(res.body).not.toBeInstanceOf(Uint8Array);
@@ -225,6 +228,7 @@ describe("createHandlerFactory()", function () {
             "Connect-Protocol-Version": "UNEXPECTED",
           }),
           body: createAsyncIterableBytes(new Uint8Array()),
+          signal: new AbortController().signal,
         });
         expect(res.status).toBe(200);
         expect(res.body).not.toBeInstanceOf(Uint8Array);
@@ -250,7 +254,7 @@ describe("createHandlerFactory()", function () {
           {},
           async (req, ctx) => {
             await new Promise((r) => setTimeout(r, timeoutMs + 50));
-            ctx.deadline?.throwIfAborted();
+            ctx.signal.throwIfAborted();
             return { value: req.value.toString(10) };
           }
         );
@@ -262,6 +266,7 @@ describe("createHandlerFactory()", function () {
           ),
           header: requestHeader(method.kind, true, timeoutMs, undefined),
           body: createAsyncIterable([new Uint8Array(0)]),
+          signal: new AbortController().signal,
         });
         expect(res.status).toBe(408);
         expect(res.body).toBeDefined();
@@ -306,7 +311,7 @@ describe("createHandlerFactory()", function () {
           {},
           async function* (req, ctx) {
             await new Promise((r) => setTimeout(r, timeoutMs + 50));
-            ctx.deadline?.throwIfAborted();
+            ctx.signal.throwIfAborted();
             yield { value: req.value.toString(10) };
           }
         );
@@ -318,6 +323,7 @@ describe("createHandlerFactory()", function () {
           ),
           header: requestHeader(method.kind, true, timeoutMs, undefined),
           body: createAsyncIterable([encodeEnvelope(0, new Uint8Array(0))]),
+          signal: new AbortController().signal,
         });
         expect(res.status).toBe(200);
         expect(res.body).toBeDefined();

packages/connect/src/protocol-connect/handler-factory.ts

@@ -157,6 +157,7 @@ function createUnaryHandler<I extends Message<I>, O extends Message<O>>(
     const context = createHandlerContext(
       spec,
       deadline.signal,
+      req.signal,
       req.header,
       {
         [headerContentType]: type.binary
@@ -292,6 +293,7 @@ function createStreamHandler<I extends Message<I>, O extends Message<O>>(
     const context = createHandlerContext(
       spec,
       deadline.signal,
+      req.signal,
       req.header,
       {
         [headerContentType]: type.binary

packages/connect/src/protocol-grpc-web/handler-factory.spec.ts

@@ -152,7 +152,7 @@ describe("createHandlerFactory()", function () {
         {},
         async (req, ctx) => {
           await new Promise((r) => setTimeout(r, timeoutMs + 50));
-          ctx.deadline?.throwIfAborted();
+          ctx.signal.throwIfAborted();
           return { value: req.value.toString(10) };
         }
       );
@@ -162,6 +162,7 @@ describe("createHandlerFactory()", function () {
         url: new URL(`https://example.com/${service.typeName}/${method.name}`),
         header: requestHeader(true, timeoutMs, undefined),
         body: createAsyncIterable([encodeEnvelope(0, new Uint8Array(0))]),
+        signal: new AbortController().signal,
       });
       expect(res.status).toBe(200);
       const lastEnv = await getLastEnvelope(res);

packages/connect/src/protocol-grpc-web/handler-factory.ts

@@ -127,6 +127,7 @@ function createHandler<I extends Message<I>, O extends Message<O>>(
     const context = createHandlerContext(
       spec,
       deadline.signal,
+      req.signal,
       req.header,
       {
         [headerContentType]: type.binary ? contentTypeProto : contentTypeJson,

packages/connect/src/protocol-grpc/handler-factory.spec.ts

@@ -130,7 +130,7 @@ describe("createHandlerFactory()", function () {
         {},
         async (req, ctx) => {
           await new Promise((r) => setTimeout(r, timeoutMs + 50));
-          ctx.deadline?.throwIfAborted();
+          ctx.signal.throwIfAborted();
           return { value: req.value.toString(10) };
         }
       );
@@ -140,6 +140,7 @@ describe("createHandlerFactory()", function () {
         url: new URL(`https://example.com/${service.typeName}/${method.name}`),
         header: requestHeader(true, timeoutMs, undefined),
         body: createAsyncIterable([encodeEnvelope(0, new Uint8Array(0))]),
+        signal: new AbortController().signal,
       });
       expect(res.status).toBe(200);
       if (res.body instanceof Uint8Array) {

packages/connect/src/protocol-grpc/handler-factory.ts

@@ -119,6 +119,7 @@ function createHandler<I extends Message<I>, O extends Message<O>>(
     const context = createHandlerContext(
       spec,
       deadline.signal,
+      req.signal,
       req.header,
       {
         [headerContentType]: type.binary ? contentTypeProto : contentTypeJson,

packages/connect/src/protocol/universal-fetch.ts

@@ -91,6 +91,7 @@ function universalServerRequestFromFetch(
     url,
     header: req.headers,
     body: iterableFromReadableStream(req.body),
+    signal: req.signal,
   };
 }
 

packages/connect/src/protocol/universal-handler-client.ts

@@ -44,6 +44,7 @@ export function createUniversalHandlerClient(
       method: uClientReq.method,
       url: reqUrl,
       header: uClientReq.header,
+      signal: uClientReq.signal ?? new AbortController().signal,
     });
     let body = uServerRes.body ?? new Uint8Array();
     if (body instanceof Uint8Array) {

packages/connect/src/protocol/universal-handler.spec.ts

@@ -155,6 +155,7 @@ describe("negotiateProtocol()", function () {
         url: new URL("https://example.com"),
         header: new Headers({ "Content-Type": "UNSUPPORTED" }),
         body: null,
+        signal: new AbortController().signal,
       });
       expect(r.status).toBe(415);
     });
@@ -165,6 +166,7 @@ describe("negotiateProtocol()", function () {
         url: new URL("https://example.com"),
         header: new Headers({ "Content-Type": "application/x" }),
         body: null,
+        signal: new AbortController().signal,
       });
       expect(r.status).toBe(405);
     });
@@ -175,6 +177,7 @@ describe("negotiateProtocol()", function () {
         url: new URL("https://example.com"),
         header: new Headers({ "Content-Type": "application/x" }),
         body: null,
+        signal: new AbortController().signal,
       });
       expect(r.status).toBe(200);
       expect(r.header?.get("stub-handler")).toBe("1");
@@ -196,6 +199,7 @@ describe("negotiateProtocol()", function () {
           url: new URL("https://example.com"),
           header: new Headers({ "Content-Type": "application/x" }),
           body: null,
+          signal: new AbortController().signal,
         });
         expect(r.status).toBe(505);
         expect(r.header?.get("Connection")).toBe("close");
@@ -208,6 +212,7 @@ describe("negotiateProtocol()", function () {
           url: new URL("https://example.com"),
           header: new Headers({ "Content-Type": "application/x" }),
           body: null,
+          signal: new AbortController().signal,
         });
         expect(r.status).toBe(200);
         expect(r.header?.get("stub-handler")).toBe("1");

packages/connect/src/protocol/universal.ts

@@ -64,6 +64,7 @@ export interface UniversalServerRequest {
    * handled efficiently.
    */
   body: AsyncIterable<Uint8Array> | JsonValue;
+  signal: AbortSignal;
 }
 
 /**