Upgrade conformance tests to v1.0.3

packages/connect-conformance/package.json

@@ -14,7 +14,7 @@
     "connectconformance": "bin/connectconformance.cjs"
   },
   "scripts": {
-    "generate": "buf generate buf.build/connectrpc/conformance:v1.0.2",
+    "generate": "buf generate buf.build/connectrpc/conformance:v1.0.3",
     "postgenerate": "license-header src/gen",
     "prebuild": "rm -rf ./dist/*",
     "build": "npm run build:cjs && npm run build:esm",

packages/connect-conformance/src/callback-client.ts

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-import { createCallbackClient, ConnectError } from "@connectrpc/connect";
+import { createCallbackClient, ConnectError, Code } from "@connectrpc/connect";
 import type { CallbackClient, Transport } from "@connectrpc/connect";
 import {
   ClientCompatRequest,
@@ -84,9 +84,34 @@ async function unary(
 
   await wait(req.requestDelayMs);
   return new Promise<ClientResponseResult>((resolve) => {
+    const controller = new AbortController();
+    // Handles client cancellation.
+    controller.signal.addEventListener("abort", () => {
+      resolve(
+        new ClientResponseResult({
+          payloads: payloads,
+          responseHeaders: resHeaders,
+          responseTrailers: resTrailers,
+          error: convertToProtoError(
+            error ?? new ConnectError("operation aborted", Code.Canceled),
+          ),
+        }),
+      );
+    });
+    const { afterCloseSendMs } = getCancelTiming(req);
+    if (afterCloseSendMs >= 0) {
+      wait(afterCloseSendMs)
+        .then(() => controller.abort())
+        .catch(() => {});
+    }
     call(
       uReq,
       (err, uRes) => {
+        // Callback clients swallow client triggered cancellations and never
+        // call the callback. This will trigger the global error handler and fail the process.
+        if (controller.signal.aborted) {
+          throw new Error("Aborted requests should not trigger the callback");
+        }
         if (err !== undefined) {
           error = ConnectError.from(err);
           // We can't distinguish between headers and trailers here, so we just
@@ -112,6 +137,7 @@ async function unary(
       },
       {
         headers: reqHeader,
+        signal: controller.signal,
         onHeader(headers) {
           resHeaders = convertToProtoHeaders(headers);
         },
@@ -148,19 +174,30 @@ async function serverStream(
 
   await wait(req.requestDelayMs);
   return new Promise<ClientResponseResult>((resolve) => {
-    const cancelFn = client.serverStream(
+    const controller = new AbortController();
+    let abortCount = -1;
+    controller.signal.addEventListener("abort", () => {
+      abortCount = count;
+    });
+    client.serverStream(
       uReq,
       (uResp) => {
-        if (cancelTiming.afterNumResponses === 0) {
-          cancelFn();
-        }
         payloads.push(uResp.payload!);
         count++;
         if (count === cancelTiming.afterNumResponses) {
-          cancelFn();
+          controller.abort();
         }
       },
       (err) => {
+        // Callback clients swallow client triggered cancellations don't report
+        // that as an error.
+        if (
+          err === undefined &&
+          controller.signal.aborted &&
+          abortCount == count
+        ) {
+          error = new ConnectError("operation aborted", Code.Canceled);
+        }
         if (err !== undefined) {
           error = ConnectError.from(err);
           // We can't distinguish between headers and trailers here, so we just
@@ -184,6 +221,7 @@ async function serverStream(
       },
       {
         headers: reqHeader,
+        signal: controller.signal,
         onHeader(headers) {
           resHeaders = convertToProtoHeaders(headers);
         },
@@ -192,6 +230,11 @@ async function serverStream(
         },
       },
     );
+    if (cancelTiming.afterCloseSendMs >= 0) {
+      wait(cancelTiming.afterCloseSendMs)
+        .then(() => controller.abort())
+        .catch(() => {});
+    }
   });
 }
 

packages/connect-conformance/src/conformance.ts

@@ -40,7 +40,7 @@ export async function run() {
   const { archive, bin } = getArtifactNameForEnv();
   const tempDir = getTempDir();
   const binPath = joinPath(tempDir, bin);
-  if (!existsSync(binPath)) {
+  if (!existsAndMatchesVersion(binPath)) {
     const archivePath = joinPath(tempDir, archive);
     await download(`${downloadUrl}/${archive}`, archivePath);
     await extractBin(archivePath, binPath);
@@ -50,6 +50,13 @@ export async function run() {
   });
 }
 
+function existsAndMatchesVersion(bin: string) {
+  return (
+    existsSync(bin) &&
+    execFileSync(bin, ["--version"]).toString().endsWith(version)
+  );
+}
+
 async function download(url: string, path: string) {
   if (existsSync(path)) {
     return;

packages/connect-conformance/src/promise-client.ts

@@ -89,9 +89,17 @@ async function unary(
     if (idempotent) {
       call = client.idempotentUnary;
     }
+    const controller = new AbortController();
     await wait(req.requestDelayMs);
+    const { afterCloseSendMs } = getCancelTiming(req);
+    if (afterCloseSendMs >= 0) {
+      wait(afterCloseSendMs)
+        .then(() => controller.abort())
+        .catch(() => {});
+    }
     const uRes = await call(uReq, {
       headers: reqHeader,
+      signal: controller.signal,
       onHeader(headers) {
         resHeaders = convertToProtoHeaders(headers);
       },
@@ -154,7 +162,8 @@ async function serverStream(
         resTrailers = convertToProtoHeaders(trailers);
       },
     });
-    if (cancelTiming.afterNumResponses == 0) {
+    if (cancelTiming.afterCloseSendMs >= 0) {
+      await wait(cancelTiming.afterCloseSendMs);
       controller.abort();
     }
     let count = 0;
@@ -290,11 +299,11 @@ async function bidiStream(client: ConformanceClient, req: ClientCompatRequest) {
         if (next.done === true) {
           continue;
         }
+        payloads.push(next.value.payload!);
         recvCount++;
         if (cancelTiming.afterNumResponses === recvCount) {
           controller.abort();
         }
-        payloads.push(next.value.payload!);
       }
     }
     if (cancelTiming.beforeCloseSend !== undefined) {
@@ -315,11 +324,11 @@ async function bidiStream(client: ConformanceClient, req: ClientCompatRequest) {
       if (next.done === true) {
         break;
       }
+      payloads.push(next.value.payload!);
       recvCount++;
       if (cancelTiming.afterNumResponses === recvCount) {
         controller.abort();
       }
-      payloads.push(next.value.payload!);
     }
   } catch (e) {
     error = ConnectError.from(e);

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,643 b |  66,478 b |   16,380 b |
-| Connect-ES     |    4 |   168,085 b |  72,418 b |   16,852 b |
-| Connect-ES     |    8 |   193,398 b |  82,142 b |   17,475 b |
-| Connect-ES     |   16 |   227,037 b |  96,408 b |   18,237 b |
+| Connect-ES     |    1 |   152,781 b |  66,543 b |   16,402 b |
+| Connect-ES     |    4 |   168,223 b |  72,483 b |   16,877 b |
+| Connect-ES     |    8 |   193,536 b |  82,208 b |   17,489 b |
+| Connect-ES     |   16 |   227,175 b |  96,471 b |   18,238 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-web/package.json

@@ -14,13 +14,13 @@
     "build:esm": "tsc --project tsconfig.build.json --outDir ./dist/esm --declaration --declarationDir ./dist/esm",
     "conformance:safari": "npm run conformance:safari:promise && npm run conformance:client:safari:callback",
     "conformance:safari:promise": "connectconformance --mode client --conf conformance/conformance-web.yaml -- tsx conformance/client.ts --browser safari",
-    "conformance:safari:callback": "connectconformance --mode client --conf conformance/conformance-web.yaml --known-failing @conformance/known-failing-callback-client.txt -- tsx conformance/client.ts --browser safari --useCallbackClient",
+    "conformance:safari:callback": "connectconformance --mode client --conf conformance/conformance-web.yaml -- tsx conformance/client.ts --browser safari --useCallbackClient",
     "conformance:chrome:promise": "connectconformance --mode client --conf conformance/conformance-web.yaml -- tsx conformance/client.ts --browser chrome",
-    "conformance:chrome:callback": "connectconformance --mode client --conf conformance/conformance-web.yaml --known-failing @conformance/known-failing-callback-client.txt -- tsx conformance/client.ts --browser chrome --useCallbackClient",
+    "conformance:chrome:callback": "connectconformance --mode client --conf conformance/conformance-web.yaml -- tsx conformance/client.ts --browser chrome --useCallbackClient",
     "conformance:firefox:promise": "connectconformance --mode client --conf conformance/conformance-web.yaml -- tsx conformance/client.ts --browser firefox",
-    "conformance:firefox:callback": "connectconformance --mode client --conf conformance/conformance-web.yaml --known-failing @conformance/known-failing-callback-client.txt -- tsx conformance/client.ts --browser firefox --useCallbackClient",
+    "conformance:firefox:callback": "connectconformance --mode client --conf conformance/conformance-web.yaml -- tsx conformance/client.ts --browser firefox --useCallbackClient",
     "conformance:node:promise": "connectconformance --mode client --conf conformance/conformance-web-node.yaml -- tsx conformance/client.ts --browser node",
-    "conformance:node:callback": "connectconformance --mode client --conf conformance/conformance-web-node.yaml --known-failing @conformance/known-failing-callback-client.txt -- tsx conformance/client.ts --browser node --useCallbackClient",
+    "conformance:node:callback": "connectconformance --mode client --conf conformance/conformance-web-node.yaml -- tsx conformance/client.ts --browser node --useCallbackClient",
     "test": "jasmine --config=jasmine.json",
     "generate": "buf generate --template browserstack/buf.gen.yaml",
     "postgenerate": "license-header browserstack/gen",

packages/connect-web/src/connect-transport.ts

@@ -266,6 +266,7 @@ export function createConnectTransport(
         body: ReadableStream<Uint8Array>,
         trailerTarget: Headers,
         header: Headers,
+        signal: AbortSignal,
       ) {
         const reader = createEnvelopeReadableStream(body).getReader();
         let endStreamReceived = false;
@@ -298,6 +299,14 @@ export function createConnectTransport(
           }
           yield parse(data);
         }
+        // Node wil not throw an AbortError on `read` if the
+        // signal is aborted before `getReader` is called.
+        // As a work around we check at the end and throw.
+        //
+        // Ref: https://github.com/nodejs/undici/issues/1940
+        if (signal.aborted) {
+          throw new ConnectError(`${signal.reason}`, Code.Canceled);
+        }
         if (!endStreamReceived) {
           throw "missing EndStreamResponse";
         }
@@ -369,7 +378,12 @@ export function createConnectTransport(
             ...req,
             header: fRes.headers,
             trailer,
-            message: parseResponseBody(fRes.body, trailer, fRes.headers),
+            message: parseResponseBody(
+              fRes.body,
+              trailer,
+              fRes.headers,
+              req.signal,
+            ),
           };
           return res;
         },