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

.vscode/settings.json

@@ -27,6 +27,6 @@
   "deno.enablePaths": ["packages/deno/test"],
   "editor.defaultFormatter": "biomejs.biome",
   "[typescript]": {
-    "editor.defaultFormatter": "biomejs.biome"
+    "editor.defaultFormatter": "vscode.typescript-language-features"
   }
 }

packages/core/src/types-hoist/index.ts

@@ -124,10 +124,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,
@@ -147,7 +144,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,

packages/core/src/types-hoist/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;
@@ -51,13 +51,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;
   // This is to be consumed by the HttpClient integration
   virtualError?: unknown;

packages/core/src/types-hoist/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/core/src/types-hoist/whatwg/index.ts

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

packages/core/src/types-hoist/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)
+}

packages/core/src/utils-hoist/instrument/fetch.ts

@@ -1,5 +1,5 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
-import type { HandlerDataFetch } from '../../types-hoist';
+import type { HandlerDataFetch, WebFetchResponse, WebReadableStreamDefaultReader } from '../../types-hoist';
 
 import { isError } from '../is';
 import { addNonEnumerableProperty, fill } from '../object';
@@ -117,55 +117,102 @@ function instrumentFetch(onFetchResolved?: (response: Response) => void, skipNat
   });
 }
 
-async function resolveResponse(res: Response | undefined, onFinishedResolving: () => void): Promise<void> {
-  if (res && res.body) {
-    const body = res.body;
-    const responseReader = body.getReader();
+async function resolveReader(reader: WebReadableStreamDefaultReader, onFinishedResolving: () => void): Promise<void> {
+  let running = true;
+  while (running) {
+    try {
+      // This .read() call will reject/throw when `reader.cancel()`
+      const { done } = await reader.read();
 
-    // Define a maximum duration after which we just cancel
-    const maxFetchDurationTimeout = setTimeout(
-      () => {
-        body.cancel().then(null, () => {
-          // noop
-        });
-      },
-      90 * 1000, // 90s
-    );
-
-    let readingActive = true;
-    while (readingActive) {
-      let chunkTimeout;
-      try {
-        // abort reading if read op takes more than 5s
-        chunkTimeout = setTimeout(() => {
-          body.cancel().then(null, () => {
-            // noop on error
-          });
-        }, 5000);
-
-        // This .read() call will reject/throw when we abort due to timeouts through `body.cancel()`
-        const { done } = await responseReader.read();
-
-        clearTimeout(chunkTimeout);
+      running = !done;
 
-        if (done) {
-          onFinishedResolving();
-          readingActive = false;
-        }
-      } catch (error) {
-        readingActive = false;
-      } finally {
-        clearTimeout(chunkTimeout);
+      if (done) {
+        onFinishedResolving();
       }
+    } catch (_) {
+      running = false;
+    }
+  }
+}
+
+/**
+ * Resolves the body stream of a `Response` object and links its cancellation to a parent `Response` body.
+ *
+ * This function attaches a custom `cancel` behavior to both the parent `Response` body and its `getReader()` method.
+ * When the parent stream or its reader is canceled, it triggers the cancellation of the child stream as well.
+ * The function also monitors the resolution of the child's body stream using `resolveReader` and performs cleanup.
+ *
+ * @param {Response} res - The `Response` object whose body stream will be resolved.
+ * @param {Response} parentRes - The parent `Response` object whose body stream is linked to the cancellation of `res`.
+ * @param {() => void} onFinishedResolving - A callback function to be invoked when the body stream of `res` is fully resolved.
+ *
+ * Export For Test Only
+ */
+export function resolveResponse(
+  res: WebFetchResponse,
+  parentRes: WebFetchResponse,
+  onFinishedResolving: () => void,
+): void {
+  if (!res.body || !parentRes.body) {
+    if (res.body) {
+      res.body.cancel().catch(_ => {
+        // noop on error
+      });
     }
 
-    clearTimeout(maxFetchDurationTimeout);
+    return;
+  }
+
+  const body = res.body;
+  const parentBody = parentRes.body;
+  // According to the WHATWG Streams API specification, when a stream is locked by calling `getReader()`,
+  // invoking `stream.cancel()` will result in a TypeError.
+  // To cancel while the stream is locked, must use `reader.cancel()`
+  // @seealso: https://streams.spec.whatwg.org
+  const responseReader = body.getReader();
+
+  const originalCancel = parentBody.cancel.bind(parentBody) as (reason?: any) => Promise<any>;
 
-    responseReader.releaseLock();
-    body.cancel().then(null, () => {
+  // Override cancel method on parent response's body
+  parentBody.cancel = async (reason?: any) => {
+    responseReader.cancel('Cancelled by parent stream').catch(_ => {
       // noop on error
     });
-  }
+
+    await originalCancel(reason);
+  };
+
+  const originalGetReader = parentRes.body.getReader.bind(parentBody) as (
+    options: ReadableStreamGetReaderOptions,
+  ) => ReadableStreamDefaultReader;
+
+  // Override getReader on parent response's body
+  parentBody.getReader = ((opts?: any) => {
+    const reader = originalGetReader(opts) as ReadableStreamDefaultReader;
+
+    const originalReaderCancel = reader.cancel.bind(reader) as (reason?: any) => Promise<any>;
+
+    reader.cancel = async (reason?: any) => {
+      responseReader.cancel('Cancelled by parent reader').catch(_ => {
+        // noop on error
+      });
+
+      await originalReaderCancel(reason);
+    };
+
+    return reader;
+  }) as any;
+
+  resolveReader(responseReader, onFinishedResolving).finally(() => {
+    try {
+      responseReader.releaseLock();
+      body.cancel().catch(() => {
+        // noop on error
+      });
+    } catch (_) {
+      // noop on error
+    }
+  });
 }
 
 function streamHandler(response: Response): void {
@@ -177,8 +224,7 @@ function streamHandler(response: Response): void {
     return;
   }
 
-  // eslint-disable-next-line @typescript-eslint/no-floating-promises
-  resolveResponse(clonedResponseForResolving, () => {
+  resolveResponse(clonedResponseForResolving as WebFetchResponse, response as WebFetchResponse, () => {
     triggerHandlers('fetch-body-resolved', {
       endTimestamp: timestampInSeconds() * 1000,
       response,