feat: timeout stream reading + small refactor

Author: chargome

packages/utils/src/instrument/fetch.ts

@@ -23,7 +23,7 @@ type FetchResource = string | { toString(): string } | { url: string };
 export function addFetchInstrumentationHandler(handler: (data: HandlerDataFetch) => void): void {
   const type = 'fetch';
   addHandler(type, handler);
-  maybeInstrument(type, instrumentFetch);
+  maybeInstrument(type, () => instrumentFetch(type));
 }
 
 /**
@@ -37,18 +37,17 @@ export function addFetchInstrumentationHandler(handler: (data: HandlerDataFetch)
 export function addFetchEndInstrumentationHandler(handler: (data: HandlerDataFetch) => void): void {
   const type = 'fetch-body-resolved';
   addHandler(type, handler);
-  maybeInstrument(type, instrumentFetchBodyReceived);
+  maybeInstrument(type, () => instrumentFetch(type));
 }
 
-function instrumentFetch(): void {
+function instrumentFetch(handlerType: 'fetch' | 'fetch-body-resolved'): void {
   if (!supportsNativeFetch()) {
     return;
   }
 
   fill(GLOBAL_OBJ, 'fetch', function (originalFetch: () => void): () => void {
     return function (...args: any[]): void {
       const { method, url } = parseFetchArgs(args);
-
       const handlerData: HandlerDataFetch = {
         args,
         fetchData: {
@@ -58,9 +57,11 @@ function instrumentFetch(): void {
         startTimestamp: timestampInSeconds() * 1000,
       };
 
-      triggerHandlers('fetch', {
-        ...handlerData,
-      });
+      if (handlerType === 'fetch') {
+        triggerHandlers('fetch', {
+          ...handlerData,
+        });
+      }
 
       // We capture the stack right here and not in the Promise error callback because Safari (and probably other
       // browsers too) will wipe the stack trace up to this point, only leaving us with this file which is useless.
@@ -73,91 +74,100 @@ function instrumentFetch(): void {
 
       // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
       return originalFetch.apply(GLOBAL_OBJ, args).then(
-        (response: Response) => {
-          const finishedHandlerData: HandlerDataFetch = {
-            ...handlerData,
-            endTimestamp: timestampInSeconds() * 1000,
-            response,
-          };
-
-          triggerHandlers('fetch', finishedHandlerData);
+        async (response: Response) => {
+          if (handlerType === 'fetch-body-resolved') {
+            // clone response for awaiting stream
+            let clonedResponseForResolving: Response | undefined;
+            try {
+              clonedResponseForResolving = response.clone();
+            } catch (e) {
+              // noop
+              DEBUG_BUILD && logger.warn('Failed to clone response body.');
+            }
+
+            await resolveResponse(clonedResponseForResolving, () => {
+              triggerHandlers('fetch-body-resolved', {
+                endTimestamp: timestampInSeconds() * 1000,
+                response,
+              });
+            });
+          } else {
+            const finishedHandlerData: HandlerDataFetch = {
+              ...handlerData,
+              endTimestamp: timestampInSeconds() * 1000,
+              response,
+            };
+            triggerHandlers('fetch', finishedHandlerData);
+          }
+
           return response;
         },
         (error: Error) => {
-          const erroredHandlerData: HandlerDataFetch = {
-            ...handlerData,
-            endTimestamp: timestampInSeconds() * 1000,
-            error,
-          };
-
-          triggerHandlers('fetch', erroredHandlerData);
+          if (handlerType === 'fetch') {
+            const erroredHandlerData: HandlerDataFetch = {
+              ...handlerData,
+              endTimestamp: timestampInSeconds() * 1000,
+              error,
+            };
+
+            triggerHandlers('fetch', erroredHandlerData);
+
+            if (isError(error) && error.stack === undefined) {
+              // NOTE: If you are a Sentry user, and you are seeing this stack frame,
+              //       it means the error, that was caused by your fetch call did not
+              //       have a stack trace, so the SDK backfilled the stack trace so
+              //       you can see which fetch call failed.
+              error.stack = virtualStackTrace;
+              addNonEnumerableProperty(error, 'framesToPop', 1);
+            }
 
-          if (isError(error) && error.stack === undefined) {
             // NOTE: If you are a Sentry user, and you are seeing this stack frame,
-            //       it means the error, that was caused by your fetch call did not
-            //       have a stack trace, so the SDK backfilled the stack trace so
-            //       you can see which fetch call failed.
-            error.stack = virtualStackTrace;
-            addNonEnumerableProperty(error, 'framesToPop', 1);
+            //       it means the sentry.javascript SDK caught an error invoking your application code.
+            //       This is expected behavior and NOT indicative of a bug with sentry.javascript.
+            throw error;
           }
-
-          // NOTE: If you are a Sentry user, and you are seeing this stack frame,
-          //       it means the sentry.javascript SDK caught an error invoking your application code.
-          //       This is expected behavior and NOT indicative of a bug with sentry.javascript.
-          throw error;
         },
       );
     };
   });
 }
 
-function instrumentFetchBodyReceived(): void {
-  if (!supportsNativeFetch()) {
-    return;
-  }
+function resolveResponse(res: Response | undefined, onFinishedResolving: () => void): void {
+  if (res && res.body) {
+    const responseReader = res.body.getReader();
 
-  fill(GLOBAL_OBJ, 'fetch', function (originalFetch: () => void): () => void {
-    return function (...args: any[]): void {
-      // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-      return originalFetch.apply(GLOBAL_OBJ, args).then(async (response: Response) => {
-        // clone response for awaiting stream
-        let clonedResponseForResolving: Response | undefined;
+    // eslint-disable-next-line no-inner-declarations
+    async function consumeChunks({ done }: { done: boolean }): Promise<void> {
+      if (!done) {
         try {
-          clonedResponseForResolving = response.clone();
-        } catch (e) {
-          // noop
-          DEBUG_BUILD && logger.warn('Failed to clone response body.');
-        }
-
-        if (clonedResponseForResolving && clonedResponseForResolving.body) {
-          const responseReader = clonedResponseForResolving.body.getReader();
-
-          // eslint-disable-next-line no-inner-declarations
-          function consumeChunks({ done }: { done: boolean }): Promise<void> {
-            if (!done) {
-              return responseReader.read().then(consumeChunks);
-            } else {
-              return Promise.resolve();
-            }
-          }
-
-          responseReader
-            .read()
-            .then(consumeChunks)
-            .then(() => {
-              triggerHandlers('fetch-body-resolved', {
-                endTimestamp: timestampInSeconds() * 1000,
-                response,
-              });
-            })
-            .catch(() => {
-              // noop
-            });
+          // abort reading if read op takes more than 5s
+          const result = await Promise.race([
+            responseReader.read(),
+            new Promise<{ done: boolean }>(res => {
+              setTimeout(() => {
+                res({ done: true });
+              }, 5000);
+            }),
+          ]);
+          await consumeChunks(result);
+        } catch (error) {
+          // handle error if needed
         }
-        return response;
+      } else {
+        return Promise.resolve();
+      }
+    }
+
+    responseReader
+      .read()
+      .then(consumeChunks)
+      .then(() => {
+        onFinishedResolving();
+      })
+      .catch(() => {
+        // noop
       });
-    };
-  });
+  }
 }
 
 function hasProp<T extends string>(obj: unknown, prop: T): obj is Record<string, string> {