Try catch subscription handlers

.changeset/bright-squids-destroy.md

@@ -0,0 +1,6 @@
+---
+"@osdk/client": patch
+"@osdk/api": patch
+---
+
+Try-catches handlers called during subscription

etc/api.report.api.md

@@ -595,7 +595,10 @@ export interface ObjectSet<Q extends ObjectOrInterfaceDefinition = any, _UNUSED
 export interface ObjectSetListener<O extends ObjectOrInterfaceDefinition, P extends PropertyKeys<O> = PropertyKeys<O>> {
     // Warning: (ae-forgotten-export) The symbol "ObjectUpdate" needs to be exported by the entry point index.d.ts
     onChange?: (objectUpdate: ObjectUpdate<O, P>) => void;
-    onError?: (errors: Array<any>) => void;
+    onError?: (errors: {
+        subscriptionClosed: boolean;
+        error: any;
+    }) => void;
     onOutOfDate?: () => void;
     onSuccessfulSubscription?: () => void;
 }

packages/api/src/objectSet/ObjectSetListener.ts

@@ -45,7 +45,7 @@ export interface ObjectSetListener<
   /**
    * There was a fatal error with the subscription process. The subscription will close or will not be established.
    */
-  onError?: (errors: Array<any>) => void;
+  onError?: (errors: { subscriptionClosed: boolean; error: any }) => void;
 }
 
 type ObjectUpdate<

packages/client/src/objectSet/ObjectSetListenerWebsocket.test.ts

@@ -258,6 +258,25 @@ describe("ObjectSetListenerWebsocket", async () => {
         });
       });
 
+      describe("correctly try catches errors in handlers", () => {
+        beforeEach(() => {
+          listener.onSuccessfulSubscription.mockImplementationOnce(() => {
+            throw new Error("I am an error");
+          });
+          respondSuccessToSubscribe(ws, subReq1);
+        });
+        afterEach(() => {
+          listener.onSuccessfulSubscription.mockReset();
+        });
+
+        it("should call onError", async () => {
+          expect(listener.onError).toHaveBeenCalled();
+          expect(listener.onError.mock.calls[0][0].subscriptionClosed).toBe(
+            false,
+          );
+        });
+      });
+
       describe("successfully subscribed", () => {
         beforeEach(() => {
           respondSuccessToSubscribe(ws, subReq1);

packages/client/src/objectSet/ObjectSetListenerWebsocket.ts

@@ -240,7 +240,7 @@ export class ObjectSetListenerWebsocket {
       }
     } catch (error) {
       this.#logger?.error(error, "Error in #initiateSubscribe");
-      sub.listener.onError([error]);
+      sub.listener.onError({ subscriptionClosed: true, error });
     }
   }
 
@@ -454,7 +454,12 @@ export class ObjectSetListenerWebsocket {
 
     for (const osdkObject of osdkObjectsWithReferenceUpdates) {
       if (osdkObject != null) {
-        sub.listener.onChange?.(osdkObject);
+        try {
+          sub.listener.onChange?.(osdkObject);
+        } catch (error) {
+          this.#logger?.error(error, "Error in onChange callback");
+          sub.listener.onError({ subscriptionClosed: false, error });
+        }
       }
     }
 
@@ -482,14 +487,25 @@ export class ObjectSetListenerWebsocket {
 
     for (const osdkObject of osdkObjects) {
       if (osdkObject != null) {
-        sub.listener.onChange?.(osdkObject);
+        try {
+          sub.listener.onChange?.(osdkObject);
+        } catch (error) {
+          this.#logger?.error(error, "Error in onChange callback");
+          sub.listener.onError({ subscriptionClosed: false, error });
+        }
       }
     }
   };
 
   #handleMessage_refreshObjectSet = (payload: RefreshObjectSet) => {
     const sub = this.#subscriptions.get(payload.id);
     invariant(sub, `Expected subscription id ${payload.id}`);
+    try {
+      sub.listener.onOutOfDate();
+    } catch (error) {
+      this.#logger?.error(error, "Error in onOutOfDate callback");
+      sub.listener.onError({ subscriptionClosed: false, error });
+    }
     sub.listener.onOutOfDate();
   };
 
@@ -508,7 +524,10 @@ export class ObjectSetListenerWebsocket {
 
       switch (response.type) {
         case "error":
-          sub.listener.onError(response.errors);
+          sub.listener.onError({
+            subscriptionClosed: true,
+            error: response.errors,
+          });
           this.#unsubscribe(sub, "error");
           break;
 
@@ -532,20 +551,27 @@ export class ObjectSetListenerWebsocket {
             sub.subscriptionId = response.id;
             this.#subscriptions.set(sub.subscriptionId, sub); // future messages come by this subId
           }
-          if (shouldFireOutOfDate) sub.listener.onOutOfDate();
-          else sub.listener.onSuccessfulSubscription();
+          try {
+            if (shouldFireOutOfDate) sub.listener.onOutOfDate();
+            else sub.listener.onSuccessfulSubscription();
+          } catch (error) {
+            this.#logger?.error(
+              error,
+              "Error in onOutOfDate or onSuccessfulSubscription callback",
+            );
+            sub.listener.onError({ subscriptionClosed: false, error });
+          }
           break;
         default:
-          const _: never = response;
-          sub.listener.onError(response);
+          sub.listener.onError({ error: response, subscriptionClosed: true });
       }
     }
   };
 
   #handleMessage_subscriptionClosed(payload: SubscriptionClosed) {
     const sub = this.#subscriptions.get(payload.id);
     invariant(sub, `Expected subscription id ${payload.id}`);
-    sub.listener.onError([payload.cause]);
+    sub.listener.onError({ subscriptionClosed: true, error: payload.cause });
     this.#unsubscribe(sub, "error");
   }