Throw an error on missing status in gRPC and gRPC-Web transports

packages/connect-node/src/transport.spec.ts

@@ -0,0 +1,113 @@
+// Copyright 2021-2024 The Connect Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/* eslint-disable @typescript-eslint/no-invalid-void-type */
+import { Int32Value, StringValue, MethodKind } from "@bufbuild/protobuf";
+import { useNodeServer } from "./use-node-server-helper.spec.js";
+import * as http2 from "node:http2";
+import { connectNodeAdapter } from "./connect-node-adapter.js";
+import { createPromiseClient } from "@connectrpc/connect";
+import type { Transport } from "@connectrpc/connect";
+import { createTransport as createGrpcTransport } from "@connectrpc/connect/protocol-grpc";
+import { createTransport as createGrpcWebTransport } from "@connectrpc/connect/protocol-grpc-web";
+import { validateNodeTransportOptions } from "./node-transport-options.js";
+
+const TestService = {
+  typeName: "TestService",
+  methods: {
+    server: {
+      name: "Server",
+      I: Int32Value,
+      O: StringValue,
+      kind: MethodKind.ServerStreaming,
+    },
+  },
+} as const;
+
+describe("Calls should fail with code internal on RST_STREAM no_error before trailers are received", function () {
+  let firstMessage: ReturnType<typeof createCompleter<void>>;
+  let rstStream: ReturnType<typeof createCompleter<void>>;
+  beforeEach(function () {
+    firstMessage = createCompleter<void>();
+    rstStream = createCompleter<void>();
+  });
+  const adaptor = connectNodeAdapter({
+    routes({ rpc }) {
+      rpc(TestService, TestService.methods.server, async function* () {
+        yield { value: "foo" };
+        // Notify to send rst stream after a message.
+        firstMessage.resolve();
+        // Wait for rst stream to be sent before returning.
+        // If we return early it will create a race.
+        await rstStream.promise;
+      });
+    },
+  });
+  const server = useNodeServer(() =>
+    http2.createServer((request, response) => {
+      adaptor(request, response);
+      firstMessage.promise
+        .then(() => {
+          response.stream.close(0, () => rstStream.resolve());
+        })
+        .catch(fail);
+    }),
+  );
+  async function testRstStream(transport: Transport) {
+    const client = createPromiseClient(TestService, transport);
+    const it = client.server({ value: 1 })[Symbol.asyncIterator]();
+    const first = await it.next();
+    expect(first.done).toBeFalse();
+    expect(first.value).toEqual(new StringValue({ value: "foo" }));
+    await expectAsync(it.next()).toBeRejected();
+  }
+  it("for gRPC Transport", async function () {
+    await testRstStream(
+      createGrpcTransport({
+        ...validateNodeTransportOptions({
+          httpVersion: "2",
+          baseUrl: server.getUrl(),
+        }),
+        baseUrl: server.getUrl(),
+        httpClient: server.getClient(),
+      }),
+    );
+  });
+  it("for gRPC Transport", async function () {
+    await testRstStream(
+      createGrpcWebTransport({
+        ...validateNodeTransportOptions({
+          httpVersion: "2",
+          baseUrl: server.getUrl(),
+        }),
+        baseUrl: server.getUrl(),
+        httpClient: server.getClient(),
+      }),
+    );
+  });
+});
+
+function createCompleter<T>() {
+  let resolve: (_: T | PromiseLike<T>) => void;
+  let reject: (reason?: unknown) => void;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return {
+    promise,
+    resolve: resolve!,
+    reject: reject!,
+  };
+}

packages/connect-node/src/use-node-server-helper.spec.ts

@@ -99,7 +99,7 @@ export function useNodeServer(
           client = createNodeHttpClient({
             httpVersion: "2",
             sessionProvider: (authority) => {
-              if (authority !== this.getUrl()) {
+              if (new URL(this.getUrl()).origin != new URL(authority).origin) {
                 throw new Error(
                   "client from useNodeServer() can only be used for requests against the server URL",
                 );

packages/connect-web-bench/README.md

@@ -15,10 +15,10 @@ usually do. We repeat this for an increasing number of RPCs.
 
 | code generator | RPCs | bundle size |  minified | compressed |
 | -------------- | ---: | ----------: | --------: | ---------: |
-| Connect-ES     |    1 |   152,759 b |  66,533 b |   16,438 b |
-| Connect-ES     |    4 |   168,201 b |  72,473 b |   16,896 b |
-| Connect-ES     |    8 |   193,514 b |  82,198 b |   17,477 b |
-| Connect-ES     |   16 |   227,153 b |  96,461 b |   18,226 b |
+| Connect-ES     |    1 |   152,762 b |  66,522 b |   16,383 b |
+| Connect-ES     |    4 |   168,204 b |  72,461 b |   16,895 b |
+| Connect-ES     |    8 |   193,517 b |  82,184 b |   17,500 b |
+| Connect-ES     |   16 |   227,156 b |  96,450 b |   18,264 b |
 | gRPC-Web       |    1 |   876,563 b | 548,495 b |   52,300 b |
 | gRPC-Web       |    4 |   928,964 b | 580,477 b |   54,673 b |
 | gRPC-Web       |    8 | 1,004,833 b | 628,223 b |   57,118 b |

packages/connect/src/protocol-connect/validate-response.ts

@@ -17,7 +17,11 @@ import { Code } from "../code.js";
 import { codeFromHttpStatus } from "./http-status.js";
 import { ConnectError } from "../connect-error.js";
 import { parseContentType } from "./content-type.js";
-import { headerStreamEncoding, headerUnaryEncoding } from "./headers.js";
+import {
+  headerContentType,
+  headerStreamEncoding,
+  headerUnaryEncoding,
+} from "./headers.js";
 import type { Compression } from "../protocol/compression.js";
 
 /**
@@ -38,7 +42,7 @@ export function validateResponse(
 ):
   | { isUnaryError: false; unaryError?: undefined }
   | { isUnaryError: true; unaryError: ConnectError } {
-  const mimeType = headers.get("Content-Type");
+  const mimeType = headers.get(headerContentType);
   const parsedType = parseContentType(mimeType);
   if (status !== 200) {
     const errorFromStatus = new ConnectError(

packages/connect/src/protocol-grpc/validate-response.spec.ts

@@ -38,17 +38,28 @@ describe("gRPC validateResponse()", function () {
   }
 
   it("should honor grpc-status field", function () {
-    const e = v(200, { "grpc-status": "8" });
+    const e = v(200, {
+      "grpc-status": "8",
+      "content-type": "application/grpc+proto",
+    });
     expect(e?.message).toBe("[resource_exhausted]");
   });
 
   it("should honor grpc-message field", function () {
-    const e = v(200, { "grpc-status": "8", "grpc-message": "out of space" });
+    const e = v(200, {
+      "grpc-status": "8",
+      "grpc-message": "out of space",
+      "content-type": "application/grpc+proto",
+    });
     expect(e?.message).toBe("[resource_exhausted] out of space");
   });
 
   it("should include headers as error metadata with grpc-status", function () {
-    const e = v(200, { "grpc-status": "8", Foo: "Bar" });
+    const e = v(200, {
+      "grpc-status": "8",
+      Foo: "Bar",
+      "content-type": "application/grpc+proto",
+    });
     expect(e?.metadata.get("Foo")).toBe("Bar");
   });
 
@@ -80,7 +91,10 @@ describe("gRPC validateResponse()", function () {
   it("should return foundStatus for grpc-status OK", function () {
     const { foundStatus } = validateResponse(
       200,
-      new Headers({ "grpc-status": "0" }),
+      new Headers({
+        "grpc-status": "0",
+        "content-type": "application/grpc+proto",
+      }),
     );
     expect(foundStatus).toBeTrue();
   });

packages/connect/src/protocol-grpc/validate-response.ts

@@ -16,8 +16,13 @@ import { codeFromHttpStatus } from "./http-status.js";
 import { ConnectError } from "../connect-error.js";
 import { findTrailerError } from "./trailer-status.js";
 import { Code } from "../code.js";
-import { headerEncoding, headerGrpcStatus } from "./headers.js";
+import {
+  headerContentType,
+  headerEncoding,
+  headerGrpcStatus,
+} from "./headers.js";
 import type { Compression } from "../protocol/compression.js";
+import { parseContentType } from "./content-type.js";
 
 /**
  * Validates response status and header for the gRPC protocol.
@@ -41,6 +46,14 @@ export function validateResponse(
       headers,
     );
   }
+  const mimeType = headers.get(headerContentType);
+  const parsedType = parseContentType(mimeType);
+  if (parsedType == undefined) {
+    throw new ConnectError(
+      `unsupported content type ${mimeType}`,
+      Code.Unknown,
+    );
+  }
   return {
     foundStatus: headers.has(headerGrpcStatus),
     headerError: findTrailerError(headers),

packages/connect/src/protocol-grpc/validate-trailer.ts

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+import { Code } from "../code.js";
+import { ConnectError } from "../connect-error.js";
+import { headerGrpcStatus } from "./headers.js";
 import { findTrailerError } from "./trailer-status.js";
 
 /**
@@ -28,4 +31,7 @@ export function validateTrailer(trailer: Headers, header: Headers): void {
     });
     throw err;
   }
+  if (!header.has(headerGrpcStatus) && !trailer.has(headerGrpcStatus)) {
+    throw new ConnectError("protocol error: missing status", Code.Internal);
+  }
 }