Try catch subscription handlers (#1019)

* Try catch subscription handlers * changeset * Rework on error
palantir · Dec 10, 2024 · cddc196 · cddc196
1 parent 720218d
commit cddc196
Show file tree

Hide file tree

Showing 5 changed files with 94 additions and 11 deletions.
diff --git a/.changeset/bright-squids-destroy.md b/.changeset/bright-squids-destroy.md
@@ -0,0 +1,6 @@
+---
+"@osdk/client": patch
+"@osdk/api": patch
+---
+
+Try-catches handlers called during subscription
diff --git a/etc/api.report.api.md b/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;
 }

diff --git a/packages/api/src/objectSet/ObjectSetListener.ts b/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<

diff --git a/packages/client/src/objectSet/ObjectSetListenerWebsocket.test.ts b/packages/client/src/objectSet/ObjectSetListenerWebsocket.test.ts
@@ -259,6 +259,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);

diff --git a/packages/client/src/objectSet/ObjectSetListenerWebsocket.ts b/packages/client/src/objectSet/ObjectSetListenerWebsocket.ts
@@ -262,7 +262,7 @@ export class ObjectSetListenerWebsocket {
       }
     } catch (error) {
       this.#logger?.error(error, "Error in #initiateSubscribe");
-      sub.listener.onError([error]);
+      this.#tryCatchOnError(sub, true, error);
     }
   }
 
@@ -483,7 +483,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");
+          this.#tryCatchOnError(sub, false, error);
+        }
       }
     }
 
@@ -510,14 +515,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");
+          this.#tryCatchOnError(sub, 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");
+      this.#tryCatchOnError(sub, false, error);
+    }
     sub.listener.onOutOfDate();
   };
 
@@ -536,7 +552,7 @@ export class ObjectSetListenerWebsocket {
 
       switch (response.type) {
         case "error":
-          sub.listener.onError(response.errors);
+          this.#tryCatchOnError(sub, true, response.errors);
           this.#unsubscribe(sub, "error");
           break;
 
@@ -560,20 +576,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",
+            );
+            this.#tryCatchOnError(sub, false, error);
+          }
           break;
         default:
-          const _: never = response;
-          sub.listener.onError(response);
+          this.#tryCatchOnError(sub, true, response);
       }
     }
   };
 
   #handleMessage_subscriptionClosed(payload: SubscriptionClosed) {
     const sub = this.#subscriptions.get(payload.id);
     invariant(sub, `Expected subscription id ${payload.id}`);
-    sub.listener.onError([payload.cause]);
+    this.#tryCatchOnError(sub, true, payload.cause);
     this.#unsubscribe(sub, "error");
   }
 
@@ -618,6 +641,38 @@ export class ObjectSetListenerWebsocket {
       this.#ensureWebsocket();
     }
   };
+
+  #tryCatchOnError = (
+    sub: Subscription<any, any>,
+    subscriptionClosed: boolean,
+    error: any,
+  ) => {
+    try {
+      sub.listener.onError({ subscriptionClosed: subscriptionClosed, error });
+    } catch (onErrorError) {
+      // eslint-disable-next-line no-console
+      console.error(
+        `Error encountered in an onError callback for an OSDK subscription`,
+        onErrorError,
+      );
+      // eslint-disable-next-line no-console
+      console.error(
+        `This onError call was triggered by an error in another callback`,
+        error,
+      );
+      // eslint-disable-next-line no-console
+      console.error(
+        `The subscription has been closed.`,
+        error,
+      );
+
+      if (!subscriptionClosed) {
+        this.#logger?.error(error, "Error in onError callback");
+        this.#unsubscribe(sub, "error");
+        this.#tryCatchOnError(sub, true, onErrorError);
+      }
+    }
+  };
 }
 
 /** @internal */