feat(core/eventStreams): add event stream mixin for HttpProtocols, schema-serde (#1671)

* feat(core/schema): event stream initial messages

* wip: initial message testing

* feat(core/eventStreams): test with eventStreamMarshaller

* handle eventHeader and eventPayload traits

* rename eventStreams submodule to event-streams

Author: kuhe

.changeset/calm-trees-teach.md

@@ -0,0 +1,5 @@
+---
+"@smithy/core": minor
+---
+
+CBOR protocol error handling fallbacks

packages/core/event-streams.d.ts

@@ -0,0 +1,7 @@
+/**
+ * Do not edit:
+ * This is a compatibility redirect for contexts that do not understand package.json exports field.
+ */
+declare module "@smithy/core/event-streams" {
+  export * from "@smithy/core/dist-types/submodules/event-streams/index.d";
+}

packages/core/event-streams.js

@@ -0,0 +1,6 @@
+
+/**
+ * Do not edit:
+ * This is a compatibility redirect for contexts that do not understand package.json exports field.
+ */
+module.exports = require("./dist-cjs/submodules/event-streams/index.js");

packages/core/package.json

@@ -60,6 +60,13 @@
       "node": "./dist-cjs/submodules/schema/index.js",
       "import": "./dist-es/submodules/schema/index.js",
       "require": "./dist-cjs/submodules/schema/index.js"
+    },
+    "./event-streams": {
+      "types": "./dist-types/submodules/event-streams/index.d.ts",
+      "module": "./dist-es/submodules/event-streams/index.js",
+      "node": "./dist-cjs/submodules/event-streams/index.js",
+      "import": "./dist-es/submodules/event-streams/index.js",
+      "require": "./dist-cjs/submodules/event-streams/index.js"
     }
   },
   "author": {
@@ -95,6 +102,8 @@
   "files": [
     "./cbor.d.ts",
     "./cbor.js",
+    "./event-streams.d.ts",
+    "./event-streams.js",
     "./protocols.d.ts",
     "./protocols.js",
     "./schema.d.ts",
@@ -110,6 +119,7 @@
     "directory": "packages/core"
   },
   "devDependencies": {
+    "@smithy/eventstream-serde-node": "workspace:^",
     "@types/node": "^18.11.9",
     "concurrently": "7.0.0",
     "downlevel-dts": "0.10.1",

packages/core/src/submodules/cbor/CborCodec.ts

@@ -139,7 +139,11 @@ export class CborShapeDeserializer implements ShapeDeserializer {
     return this.readValue(schema, data);
   }
 
-  private readValue(_schema: Schema, value: any): any {
+  /**
+   * Public because it's called by the protocol implementation to deserialize errors.
+   * @internal
+   */
+  public readValue(_schema: Schema, value: any): any {
     const ns = NormalizedSchema.of(_schema);
 
     if (ns.isTimestampSchema() && typeof value === "number") {

packages/core/src/submodules/cbor/SmithyRpcV2CborProtocol.spec.ts

@@ -1,7 +1,7 @@
-import { list, map, SCHEMA, struct } from "@smithy/core/schema";
+import { error, list, map, op, SCHEMA, struct, TypeRegistry } from "@smithy/core/schema";
 import { HttpRequest, HttpResponse } from "@smithy/protocol-http";
-import type { SchemaRef } from "@smithy/types";
-import { describe, expect, test as it } from "vitest";
+import { ResponseMetadata, RetryableTrait, SchemaRef } from "@smithy/types";
+import { beforeEach, describe, expect, test as it } from "vitest";
 
 import { cbor } from "./cbor";
 import { dateToTag } from "./parseCborBody";
@@ -273,4 +273,92 @@ describe(SmithyRpcV2CborProtocol.name, () => {
       });
     }
   });
+
+  describe("error handling", () => {
+    const protocol = new SmithyRpcV2CborProtocol({ defaultNamespace: "ns" });
+
+    const operation = op(
+      "ns",
+      "OperationWithModeledException",
+      {},
+      struct("ns", "Input", 0, [], []),
+      struct("ns", "Output", 0, [], [])
+    );
+
+    const errorResponse = new HttpResponse({
+      statusCode: 400,
+      headers: {},
+      body: cbor.serialize({
+        __type: "ns#ModeledException",
+        modeledProperty: "oh no",
+      }),
+    });
+
+    const serdeContext = {};
+
+    class ServiceBaseException extends Error {
+      public readonly $fault: "client" | "server" = "client";
+      public $response?: HttpResponse;
+      public $retryable?: RetryableTrait;
+      public $metadata: ResponseMetadata = {
+        httpStatusCode: 400,
+      };
+    }
+
+    class ModeledExceptionCtor extends ServiceBaseException {
+      public modeledProperty: string = "";
+    }
+
+    beforeEach(() => {
+      TypeRegistry.for("ns").destroy();
+    });
+
+    it("should throw the schema error ctor if one exists", async () => {
+      // this is for modeled exceptions.
+
+      TypeRegistry.for("ns").register(
+        "ns#ModeledException",
+        error("ns", "ModeledException", 0, ["modeledProperty"], [0], ModeledExceptionCtor)
+      );
+      TypeRegistry.for("ns").register(
+        "smithy.ts.sdk.synthetic.ns#BaseServiceException",
+        error("smithy.ts.sdk.synthetic.ns", "BaseServiceException", 0, [], [], ServiceBaseException)
+      );
+
+      try {
+        await protocol.deserializeResponse(operation, serdeContext as any, errorResponse);
+      } catch (e) {
+        expect(e).toBeInstanceOf(ModeledExceptionCtor);
+        expect((e as ModeledExceptionCtor).modeledProperty).toEqual("oh no");
+        expect(e).toBeInstanceOf(ServiceBaseException);
+      }
+      expect.assertions(3);
+    });
+
+    it("should throw a base error if available in the namespace, when no error schema is modeled", async () => {
+      // this is the expected fallback case for all generated clients.
+
+      TypeRegistry.for("ns").register(
+        "smithy.ts.sdk.synthetic.ns#BaseServiceException",
+        error("smithy.ts.sdk.synthetic.ns", "BaseServiceException", 0, [], [], ServiceBaseException)
+      );
+
+      try {
+        await protocol.deserializeResponse(operation, serdeContext as any, errorResponse);
+      } catch (e) {
+        expect(e).toBeInstanceOf(ServiceBaseException);
+      }
+      expect.assertions(1);
+    });
+
+    it("should fall back to a generic JS Error as a last resort", async () => {
+      // this shouldn't happen, but in case the type registry is mutated incorrectly.
+      try {
+        await protocol.deserializeResponse(operation, serdeContext as any, errorResponse);
+      } catch (e) {
+        expect(e).toBeInstanceOf(Error);
+      }
+      expect.assertions(1);
+    });
+  });
 });

packages/core/src/submodules/cbor/SmithyRpcV2CborProtocol.ts

@@ -1,5 +1,5 @@
 import { RpcProtocol } from "@smithy/core/protocols";
-import { deref, ErrorSchema, OperationSchema, TypeRegistry } from "@smithy/core/schema";
+import { deref, ErrorSchema, NormalizedSchema, OperationSchema, TypeRegistry } from "@smithy/core/schema";
 import type {
   EndpointBearer,
   HandlerExecutionContext,
@@ -87,31 +87,54 @@ export class SmithyRpcV2CborProtocol extends RpcProtocol {
     dataObject: any,
     metadata: ResponseMetadata
   ): Promise<never> {
-    const error = loadSmithyRpcV2CborErrorCode(response, dataObject) ?? "Unknown";
+    const errorName = loadSmithyRpcV2CborErrorCode(response, dataObject) ?? "Unknown";
 
     let namespace = this.options.defaultNamespace;
-    if (error.includes("#")) {
-      [namespace] = error.split("#");
+    if (errorName.includes("#")) {
+      [namespace] = errorName.split("#");
     }
 
+    const errorMetadata = {
+      $metadata: metadata,
+      $response: response,
+      $fault: response.statusCode <= 500 ? ("client" as const) : ("server" as const),
+    };
+
     const registry = TypeRegistry.for(namespace);
-    const errorSchema: ErrorSchema = registry.getSchema(error) as ErrorSchema;
 
-    if (!errorSchema) {
-      // TODO(schema) throw client base exception using the dataObject.
-      throw new Error("schema not found for " + error);
+    let errorSchema: ErrorSchema;
+    try {
+      errorSchema = registry.getSchema(errorName) as ErrorSchema;
+    } catch (e) {
+      if (dataObject.Message) {
+        dataObject.message = dataObject.Message;
+      }
+      const baseExceptionSchema = TypeRegistry.for("smithy.ts.sdk.synthetic." + namespace).getBaseException();
+      if (baseExceptionSchema) {
+        const ErrorCtor = baseExceptionSchema.ctor;
+        throw Object.assign(new ErrorCtor({ name: errorName }), errorMetadata, dataObject);
+      }
+      throw Object.assign(new Error(errorName), errorMetadata, dataObject);
     }
 
+    const ns = NormalizedSchema.of(errorSchema);
     const message = dataObject.message ?? dataObject.Message ?? "Unknown";
     const exception = new errorSchema.ctor(message);
-    Object.assign(exception, {
-      $metadata: metadata,
-      $response: response,
-      message,
-      ...dataObject,
-    });
 
-    throw exception;
+    const output = {} as any;
+    for (const [name, member] of ns.structIterator()) {
+      output[name] = this.deserializer.readValue(member, dataObject[name]);
+    }
+
+    throw Object.assign(
+      exception,
+      errorMetadata,
+      {
+        $fault: ns.getMergedTraits().error,
+        message,
+      },
+      output
+    );
   }
 
   protected getDefaultContentType(): string {