fix(fetch): properly release fetch during long-lived stream handling

packages/replay-internal/src/coreHandlers/util/fetchUtils.ts

@@ -293,21 +293,40 @@ function _tryCloneResponse(response: Response): Response | void {
  * Fetch can return a streaming body, that may not resolve (or not for a long time).
  * If that happens, we rather abort after a short time than keep waiting for this.
  */
-function _tryGetResponseText(response: Response): Promise<string | undefined> {
-  return new Promise((resolve, reject) => {
-    const timeout = setTimeout(() => reject(new Error('Timeout while trying to read response body')), 500);
-
-    _getResponseText(response)
-      .then(
-        txt => resolve(txt),
-        reason => reject(reason),
-      )
-      .finally(() => clearTimeout(timeout));
-  });
-}
+async function _tryGetResponseText(response: Response): Promise<string | undefined> {
+  if (!response.body) {
+    throw new Error('Response has no body');
+  }
+
+  const reader = response.body.getReader();
+
+  const decoder = new TextDecoder();
+  let result = '';
+  let running = true;
+
+  const timeout = setTimeout(() => {
+    if (running) {
+      reader.cancel('Timeout while trying to read response body').catch(_ => {
+        // This block is only triggered when stream has already been released,
+        // so it's safe to ignore.
+      });
+    }
+  }, 500);
 
-async function _getResponseText(response: Response): Promise<string> {
-  // Force this to be a promise, just to be safe
-  // eslint-disable-next-line no-return-await
-  return await response.text();
+  try {
+    while (running) {
+      const { value, done } = await reader.read();
+
+      running = !done;
+
+      if (value) {
+        result += decoder.decode(value, { stream: !done });
+      }
+    }
+  } finally {
+    clearTimeout(timeout);
+    reader.releaseLock();
+  }
+
+  return result;
 }

packages/replay-internal/test/unit/coreHandlers/handleNetworkBreadcrumbs.test.ts

@@ -40,6 +40,8 @@ function getMockResponse(contentLength?: string, body?: string, headers?: Record
     ...headers,
   };
 
+  const encoder = new TextEncoder();
+
   const response = {
     headers: {
       has: (prop: string) => {
@@ -49,6 +51,24 @@ function getMockResponse(contentLength?: string, body?: string, headers?: Record
         return internalHeaders[prop.toLowerCase() ?? ''];
       },
     },
+    body: {
+      getReader: () => {
+        return {
+          read: () => {
+            return Promise.resolve({
+              done: true,
+              value: encoder.encode(body),
+            });
+          },
+          cancel: async () => {
+            // noop
+          },
+          releaseLock: async () => {
+            // noop
+          },
+        };
+      },
+    },
     clone: () => response,
     text: () => Promise.resolve(body),
   } as unknown as Response;
@@ -741,6 +761,7 @@ other-header: test`;
       options.networkCaptureBodies = true;
 
       const largeBody = JSON.stringify({ a: LARGE_BODY });
+      const encoder = new TextEncoder();
 
       const breadcrumb: Breadcrumb = {
         category: 'fetch',
@@ -756,6 +777,24 @@ other-header: test`;
           get: () => '',
         },
         clone: () => mockResponse,
+        body: {
+          getReader: () => {
+            return {
+              read: () => {
+                return Promise.resolve({
+                  done: true,
+                  value: encoder.encode(largeBody),
+                });
+              },
+              cancel: async () => {
+                // noop
+              },
+              releaseLock: async () => {
+                // noop
+              },
+            };
+          },
+        },
         text: () => Promise.resolve(largeBody),
       } as unknown as Response;
 

packages/replay-internal/test/unit/coreHandlers/util/fetchUtils.test.ts

@@ -43,6 +43,19 @@ describe('Unit | coreHandlers | util | fetchUtils', () => {
     });
 
     it('works with text body', async () => {
+      const encoder = new TextEncoder();
+
+      const mockRead = vi
+        .fn()
+        .mockResolvedValueOnce({
+          value: encoder.encode('text body'),
+          done: false,
+        })
+        .mockResolvedValueOnce({
+          value: null,
+          done: true,
+        });
+
       const response = {
         headers: {
           has: () => {
@@ -52,6 +65,19 @@ describe('Unit | coreHandlers | util | fetchUtils', () => {
             return undefined;
           },
         },
+        body: {
+          getReader: () => {
+            return {
+              read: mockRead,
+              cancel: async (reason?: any) => {
+                mockRead.mockRejectedValue(new Error(reason));
+              },
+              releaseLock: async () => {
+                // noop
+              },
+            };
+          },
+        },
         clone: () => response,
         text: () => Promise.resolve('text body'),
       } as unknown as Response;
@@ -74,6 +100,8 @@ describe('Unit | coreHandlers | util | fetchUtils', () => {
     });
 
     it('works with body that fails', async () => {
+      const mockRead = vi.fn().mockRejectedValueOnce(new Error('cannot read'));
+
       const response = {
         headers: {
           has: () => {
@@ -83,6 +111,19 @@ describe('Unit | coreHandlers | util | fetchUtils', () => {
             return undefined;
           },
         },
+        body: {
+          getReader: () => {
+            return {
+              read: mockRead,
+              cancel: async (_?: any) => {
+                // noop
+              },
+              releaseLock: async () => {
+                // noop
+              },
+            };
+          },
+        },
         clone: () => response,
         text: () => Promise.reject('cannot read'),
       } as unknown as Response;
@@ -105,6 +146,12 @@ describe('Unit | coreHandlers | util | fetchUtils', () => {
     });
 
     it('works with body that times out', async () => {
+      const encoder = new TextEncoder();
+      const mockRead = vi.fn();
+
+      let cancelled = false;
+      let cancellReason = '';
+
       const response = {
         headers: {
           has: () => {
@@ -114,6 +161,38 @@ describe('Unit | coreHandlers | util | fetchUtils', () => {
             return undefined;
           },
         },
+        body: {
+          getReader: () => {
+            return {
+              read: async () => {
+                if (cancelled) {
+                  mockRead.mockRejectedValue(new Error(cancellReason));
+                } else {
+                  mockRead.mockResolvedValueOnce({
+                    value: encoder.encode('chunk'),
+                    done: false,
+                  });
+                }
+
+                await new Promise(res => {
+                  setTimeout(() => {
+                    res(1);
+                  }, 200);
+                });
+
+                // eslint-disable-next-line no-return-await
+                return await mockRead();
+              },
+              cancel: async (reason?: any) => {
+                cancelled = true;
+                cancellReason = reason;
+              },
+              releaseLock: async () => {
+                // noop
+              },
+            };
+          },
+        },
         clone: () => response,
         text: () => new Promise(resolve => setTimeout(() => resolve('text body'), 1000)),
       } as unknown as Response;

packages/types/src/index.ts

@@ -123,10 +123,7 @@ export type { StackFrame } from './stackframe';
 export type { Stacktrace, StackParser, StackLineParser, StackLineParserFn } from './stacktrace';
 export type { PropagationContext, TracePropagationTargets, SerializedTraceData } from './tracing';
 export type { StartSpanOptions } from './startSpanOptions';
-export type {
-  TraceparentData,
-  TransactionSource,
-} from './transaction';
+export type { TraceparentData, TransactionSource } from './transaction';
 export type { CustomSamplingContext, SamplingContext } from './samplingcontext';
 export type {
   DurationUnit,
@@ -146,7 +143,14 @@ export type {
   TransportRequestExecutor,
 } from './transport';
 export type { User } from './user';
-export type { WebFetchHeaders, WebFetchRequest } from './webfetchapi';
+export type {
+  WebFetchHeaders,
+  WebFetchRequest,
+  WebFetchResponse,
+  WebReadableStream,
+  WebReadableStreamDefaultReader,
+  WebReadableStreamReadResult,
+} from './whatwg';
 export type { WrappedFunction } from './wrappedfunction';
 export type {
   HandlerDataFetch,
@@ -163,13 +167,7 @@ export type {
 
 export type { BrowserClientReplayOptions, BrowserClientProfilingOptions } from './browseroptions';
 export type { CheckIn, MonitorConfig, FinishedCheckIn, InProgressCheckIn, SerializedCheckIn } from './checkin';
-export type {
-  MetricsAggregator,
-  MetricBucketItem,
-  MetricInstance,
-  MetricData,
-  Metrics,
-} from './metrics';
+export type { MetricsAggregator, MetricBucketItem, MetricInstance, MetricData, Metrics } from './metrics';
 export type { ParameterizedString } from './parameterize';
 export type { ContinuousProfiler, ProfilingIntegration, Profiler } from './profiling';
 export type { ViewHierarchyData, ViewHierarchyWindow } from './view-hierarchy';

packages/types/src/instrument.ts

@@ -1,7 +1,7 @@
 // This should be: null | Blob | BufferSource | FormData | URLSearchParams | string
 // But since not all of those are available in node, we just export `unknown` here for now
 
-import type { WebFetchHeaders } from './webfetchapi';
+import type { WebFetchResponse } from './whatwg';
 
 // Make sure to cast it where needed!
 type XHRSendInput = unknown;
@@ -48,13 +48,7 @@ export interface HandlerDataFetch {
   fetchData: SentryFetchData; // This data is among other things dumped directly onto the fetch breadcrumb data
   startTimestamp: number;
   endTimestamp?: number;
-  // This is actually `Response` - Note: this type is not complete. Add to it if necessary.
-  response?: {
-    readonly ok: boolean;
-    readonly status: number;
-    readonly url: string;
-    headers: WebFetchHeaders;
-  };
+  response?: WebFetchResponse;
   error?: unknown;
 }
 

packages/types/src/whatwg/fetch.ts

@@ -0,0 +1,38 @@
+// These are vendored types for the standard web fetch API types because typescript needs the DOM types to be able to understand the `Request`, `Headers`, ... types and not everybody has those.
+
+import type { WebReadableStream } from './stream';
+
+export interface WebFetchHeaders {
+  append(name: string, value: string): void;
+  delete(name: string): void;
+  get(name: string): string | null;
+  has(name: string): boolean;
+  set(name: string, value: string): void;
+  forEach(callbackfn: (value: string, key: string, parent: WebFetchHeaders) => void): void;
+}
+
+export interface WebFetchRequest {
+  readonly headers: WebFetchHeaders;
+  readonly method: string;
+  readonly url: string;
+  clone(): WebFetchRequest;
+}
+
+export interface WebFetchResponse {
+  readonly ok: boolean;
+  readonly status: number;
+  readonly statusText: string;
+  readonly headers: WebFetchHeaders;
+  readonly url: string;
+  readonly redirected: boolean;
+  readonly body: WebReadableStream | null;
+
+  clone(): WebFetchResponse;
+
+  // Methods to consume the response body
+  json(): Promise<any>; // Parses response as JSON
+  text(): Promise<string>; // Reads response body as text
+  arrayBuffer(): Promise<ArrayBuffer>; // Reads response body as ArrayBuffer
+  blob(): Promise<object>; // Reads response body as Blob
+  formData(): Promise<object>; // Reads response body as FormData
+}

packages/types/src/whatwg/index.ts

@@ -0,0 +1,3 @@
+export type { WebReadableStream, WebReadableStreamDefaultReader, WebReadableStreamReadResult } from './stream';
+
+export type { WebFetchHeaders, WebFetchRequest, WebFetchResponse } from './fetch';

packages/types/src/whatwg/stream.ts

@@ -0,0 +1,23 @@
+export interface WebReadableStream<R = any> {
+  locked: boolean; // Indicates if the stream is currently locked
+
+  cancel(reason?: any): Promise<void>; // Cancels the stream with an optional reason
+  getReader(): WebReadableStreamDefaultReader<R>; // Returns a reader for the stream
+}
+
+export interface WebReadableStreamDefaultReader<R = any> {
+  closed: boolean;
+  // Closes the stream and resolves the reader's lock
+  cancel(reason?: any): Promise<void>;
+
+  // Returns a promise with the next chunk in the stream
+  read(): Promise<WebReadableStreamReadResult<R>>;
+
+  // Releases the reader's lock on the stream
+  releaseLock(): void;
+}
+
+export interface WebReadableStreamReadResult<R = any> {
+  done: boolean; // True if the reader is done with the stream
+  value?: R; // The data chunk read from the stream (if not done)
+}