catch construct error and re-report as protocol error (#353)

## Why

construct can reject and cause an otherwise uncaught async rejection

## What changed

catch construct error and re-report as protocol error

## Versioning

- [ ] Breaking protocol change
- [ ] Breaking ts/js API change

<!-- Kind reminder to add tests and updated documentation if needed -->

Author: jackyzha0

package-lock.json

@@ -1,7 +1,7 @@
 {
   "name": "@replit/river",
   "description": "It's like tRPC but... with JSON Schema Support, duplex streaming and support for service multiplexing. Transport agnostic!",
-  "version": "0.212.0",
+  "version": "0.212.1",
   "type": "module",
   "exports": {
     ".": {

package.json

@@ -480,7 +480,23 @@ export abstract class ClientTransport<
     let metadata: unknown = undefined;
 
     if (this.handshakeExtensions) {
-      metadata = await this.handshakeExtensions.construct();
+      try {
+        metadata = await this.handshakeExtensions.construct();
+      } catch (err) {
+        const errStr = coerceErrorString(err);
+        this.log?.error(
+          `failed to construct handshake metadata for session to ${session.to}: ${errStr}`,
+          session.loggingMetadata,
+        );
+
+        this.protocolError({
+          type: ProtocolError.HandshakeFailed,
+          message: `failed to construct handshake metadata: ${errStr}`,
+        });
+        this.deleteSession(session, { unhealthy: true });
+
+        return;
+      }
     }
 
     // double-check to make sure we haven't transitioned the session yet

transport/client.ts

@@ -1590,6 +1590,66 @@ describe.each(testMatrix())(
       });
     });
 
+    test('construct throwing during handshake results in protocol error', async () => {
+      const schema = Type.Object({
+        foo: Type.String(),
+      });
+
+      interface Metadata {
+        foo: string;
+      }
+
+      const constructError = new Error('Failed to construct metadata');
+      const get = vi.fn(async () => {
+        throw constructError;
+      });
+      const parse = vi.fn(async (metadata: Metadata) => ({
+        foo: metadata.foo,
+      }));
+
+      const serverTransport = getServerTransport('SERVER', {
+        schema,
+        validate: parse,
+      });
+      const clientTransport = getClientTransport('client', {
+        schema,
+        construct: get,
+      });
+      const clientHandshakeFailed = vi.fn();
+      clientTransport.addEventListener('protocolError', clientHandshakeFailed);
+      clientTransport.connect(serverTransport.clientId);
+
+      addPostTestCleanup(async () => {
+        clientTransport.removeEventListener(
+          'protocolError',
+          clientHandshakeFailed,
+        );
+        await cleanupTransports([clientTransport, serverTransport]);
+      });
+
+      await waitFor(() => {
+        expect(get).toHaveBeenCalledTimes(1);
+        expect(clientHandshakeFailed).toHaveBeenCalledTimes(1);
+        expect(clientHandshakeFailed).toHaveBeenCalledWith(
+          expect.objectContaining({
+            type: ProtocolError.HandshakeFailed,
+            message:
+              'failed to construct handshake metadata: Failed to construct metadata',
+          }),
+        );
+        // should never get to the server
+        expect(parse).toHaveBeenCalledTimes(0);
+      });
+
+      expect(numberOfConnections(clientTransport)).toBe(0);
+      expect(numberOfConnections(serverTransport)).toBe(0);
+
+      await testFinishesCleanly({
+        clientTransports: [clientTransport],
+        serverTransport,
+      });
+    });
+
     test('server checks request schema on receive', async () => {
       const clientRequestSchema = Type.Object({
         foo: Type.Number(),