fix(instrumentation-fetch): do not modify the returned type of fetch (#6521)

Author: dyladan

experimental/CHANGELOG.md

@@ -23,6 +23,7 @@ For notes on migrating to 2.x / 0.200.x see [the upgrade guide](doc/upgrade-to-2
 
 * fix(opentelemetry-instrumentation): access `require` via `globalThis` to avoid webpack analysis [#6481](https://github.com/open-telemetry/opentelemetry-js/pull/6481) @overbalance
 * fix(sdk-logs): fix inflated `droppedAttributesCount` when updating existing attribute keys [#6479](https://github.com/open-telemetry/opentelemetry-js/pull/6479) @overbalance
+* fix(instrumentation-fetch): do not modify the returned type of fetch [#6521](https://github.com/open-telemetry/opentelemetry-js/pull/6521) @dyladan
 * fix(opentelemetry-sdk-node): add missing `@opentelemetry/otlp-exporter-base` dependency [#6520](https://github.com/open-telemetry/opentelemetry-js/pull/6520) @gotgenes
 
 ### :books: Documentation

experimental/packages/opentelemetry-instrumentation-fetch/src/fetch.ts

@@ -377,27 +377,6 @@ export class FetchInstrumentation extends InstrumentationBase<FetchInstrumentati
   private _patchConstructor(): (original: typeof fetch) => typeof fetch {
     return original => {
       const plugin = this;
-      const readOnlyProps = new Set(['url', 'type', 'redirected']);
-      function createResponseProxy(
-        target: Response,
-        originalResponse: Response
-      ): Response {
-        return new Proxy(target, {
-          get(t, prop, _receiver) {
-            if (typeof prop === 'string' && readOnlyProps.has(prop)) {
-              return Reflect.get(originalResponse, prop);
-            }
-            if (prop === 'clone') {
-              return function clone() {
-                return createResponseProxy(t.clone(), originalResponse);
-              };
-            }
-            // Use target as receiver so getters (e.g. headers) run with correct this and avoid "Illegal invocation"
-            const value = Reflect.get(t, prop, t);
-            return typeof value === 'function' ? value.bind(t) : value;
-          },
-        }) as Response;
-      }
 
       return function patchConstructor(
         this: typeof globalThis,
@@ -459,81 +438,20 @@ export class FetchInstrumentation extends InstrumentationBase<FetchInstrumentati
           }
         }
 
-        function withCancelPropagation(
-          body: ReadableStream<Uint8Array> | null,
-          readerClone: ReadableStreamDefaultReader<Uint8Array>
-        ): ReadableStream<Uint8Array> | null {
-          if (!body) return null;
-
-          const reader = body.getReader();
-
-          return new ReadableStream({
-            async pull(controller) {
-              try {
-                const { value, done } = await reader.read();
-                if (done) {
-                  reader.releaseLock();
-                  controller.close();
-                } else {
-                  controller.enqueue(value);
-                }
-              } catch (err) {
-                controller.error(err);
-                reader.cancel(err).catch(_ => {});
-
-                try {
-                  reader.releaseLock();
-                } catch {
-                  // Spec reference:
-                  // https://streams.spec.whatwg.org/#default-reader-release-lock
-                  //
-                  // releaseLock() only throws if called on an invalid reader
-                  // (i.e. reader.[[stream]] is undefined, meaning the lock is already released
-                  // or the reader was never associated). In normal use this cannot happen.
-                  // This catch is defensive only.
-                }
-              }
-            },
-            cancel(reason) {
-              readerClone.cancel(reason).catch(_ => {});
-              return reader.cancel(reason);
-            },
-          });
-        }
-
-        function onSuccess(
-          span: Span,
-          resolve: (value: Response | PromiseLike<Response>) => void,
-          response: Response
-        ): void {
-          let proxiedResponse: Response | null = null;
-
+        function onSuccess(span: Span, response: Response): Response {
           try {
+            // Clone the response and eagerly consume the clone to detect
+            // when the body transfer completes.  The original response is
+            // returned to the caller untouched so that it passes internal
+            // brand-checks required by APIs such as
+            // WebAssembly.compileStreaming.
             // TODO: Switch to a consumer-driven model and drop `resClone`.
             // Keeping eager consumption here to preserve current behavior and avoid breaking existing tests.
             // Context: discussion in PR #5894 → https://github.com/open-telemetry/opentelemetry-js/pull/5894
             const resClone = response.clone();
             const body = resClone.body;
             if (body) {
               const reader = body.getReader();
-              const isNullBodyStatus =
-                // 101 responses and protocol upgrading is handled internally by the browser
-                response.status === 204 ||
-                response.status === 205 ||
-                response.status === 304;
-              const wrappedBody = isNullBodyStatus
-                ? null
-                : withCancelPropagation(response.body, reader);
-
-              const newResponse = new Response(wrappedBody, {
-                status: response.status,
-                statusText: response.statusText,
-                headers: response.headers,
-              });
-
-              // Response url, type, and redirected are read-only properties that can't be set via constructor
-              // Use a Proxy to forward them from the original response and maintain the wrapped body
-              proxiedResponse = createResponseProxy(newResponse, response);
 
               const read = (): void => {
                 reader.read().then(
@@ -554,46 +472,55 @@ export class FetchInstrumentation extends InstrumentationBase<FetchInstrumentati
               // some older browsers don't have .body implemented
               endSpanOnSuccess(span, response);
             }
-          } finally {
-            resolve(proxiedResponse ?? response);
+          } catch (error) {
+            // Setup failed (e.g. clone() or getReader() threw).
+            // End the span and clean up so _tasksCount doesn't leak.
+            plugin._diag.error('Failed to read fetch response body', error);
+            plugin._endSpan(span, spanData, {
+              status: 0,
+              url,
+            });
           }
+          return response;
         }
 
-        function onError(
-          span: Span,
-          reject: (reason?: unknown) => void,
-          error: FetchError
-        ) {
+        function onError(span: Span, error: FetchError): never {
           try {
             endSpanOnError(span, error);
-          } finally {
-            reject(error);
+          } catch (e: unknown) {
+            // endSpanOnError failed — fall back to ending the span
+            // directly so _tasksCount doesn't leak.
+            plugin._diag.error('Failed to end span on fetch error', e);
+            plugin._endSpan(span, spanData, {
+              status: error.status || 0,
+              url,
+            });
           }
+          // eslint-disable-next-line @typescript-eslint/only-throw-error
+          throw error;
         }
 
-        return new Promise((resolve, reject) => {
-          return context.with(
-            trace.setSpan(context.active(), createdSpan),
-            () => {
-              // Call request hook before injection so hooks cannot tamper with propagation headers.
-              // Also, this means the hook will see `options.headers` in the same type as passed in,
-              // rather than as a `Headers` instance set by `_addHeaders()`.
-              plugin._callRequestHook(createdSpan, options);
-              plugin._addHeaders(options, url);
-              plugin._tasksCount++;
-
-              return original
-                .apply(
-                  self,
-                  options instanceof Request ? [options] : [url, options]
-                )
-                .then(
-                  onSuccess.bind(self, createdSpan, resolve),
-                  onError.bind(self, createdSpan, reject)
-                );
-            }
-          );
-        });
+        return context.with(
+          trace.setSpan(context.active(), createdSpan),
+          () => {
+            // Call request hook before injection so hooks cannot tamper with propagation headers.
+            // Also, this means the hook will see `options.headers` in the same type as passed in,
+            // rather than as a `Headers` instance set by `_addHeaders()`.
+            plugin._callRequestHook(createdSpan, options);
+            plugin._addHeaders(options, url);
+            plugin._tasksCount++;
+
+            return original
+              .apply(
+                self,
+                options instanceof Request ? [options] : [url, options]
+              )
+              .then(
+                onSuccess.bind(self, createdSpan),
+                onError.bind(self, createdSpan)
+              );
+          }
+        );
       };
     };
   }

experimental/packages/opentelemetry-instrumentation-fetch/test/fetch.test.ts

@@ -236,6 +236,35 @@ describe('fetch', () => {
       // Avoids ERROR in the logs when calling `disable()` again during cleanup
       fetchInstrumentation = undefined;
     });
+
+    it('should return a Promise<Response> compatible with WebAssembly.compileStreaming', async () => {
+      // Some web APIs do brand checks to ensure they are working with native objects.
+      // compileStreaming checks that the argument is a native Response, and will throw if it isn't.
+
+      // Minimal valid WASM binary: magic number + version only.
+      const wasmBytes = new Uint8Array([
+        0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00,
+      ]);
+
+      await startWorker(
+        msw.http.get(`${ORIGIN}/test.wasm`, () => {
+          return new msw.HttpResponse(wasmBytes, {
+            headers: { 'Content-Type': 'application/wasm' },
+          });
+        })
+      );
+
+      fetchInstrumentation = new FetchInstrumentation();
+      assert.ok(isWrapped(window.fetch));
+
+      // WebAssembly.compileStreaming requires a native Response from fetch.
+      const module = await WebAssembly.compileStreaming(
+        fetch(`${ORIGIN}/test.wasm`)
+      );
+      assert.ok(module instanceof WebAssembly.Module);
+
+      await waitFor(OBSERVER_WAIT_TIME_MS + 50);
+    });
   });
 
   describe('instrumentation', () => {
@@ -2764,13 +2793,47 @@ describe('fetch', () => {
       });
 
       describe('when client cancels the reader', () => {
-        it('should cancel stream and release the connection', async () => {
-          const { response } = await tracedFetch();
+        it('should end the span when the stream completes', async () => {
+          // Use a short stream (3 chunks) so the clone finishes quickly
+          // after the consumer cancels.  The instrumentation tracks
+          // completion via an eagerly-consumed clone; consumer-side
+          // cancellation does not propagate to the clone.
+          const shortStreamHandler = () => {
+            const encoder = new TextEncoder();
+            return msw.http.get('/api/stream', () => {
+              let count = 0;
+              const stream = new ReadableStream<Uint8Array>({
+                start(controller) {
+                  timer = setInterval(() => {
+                    if (count >= 3) {
+                      clearInterval(timer);
+                      controller.close();
+                      return;
+                    }
+                    count += 1;
+                    controller.enqueue(encoder.encode(`data: ${count}\n`));
+                  }, 20);
+                },
+              });
+              return new msw.HttpResponse(stream, {
+                status: 200,
+                headers: {
+                  'Content-Type': 'text/plain; charset=utf-8',
+                  'Cache-Control': 'no-cache',
+                  Connection: 'keep-alive',
+                },
+              });
+            });
+          };
+
+          const { response } = await tracedFetch({
+            handlers: [shortStreamHandler()],
+          });
 
           const timeoutPromise = new Promise((_, reject) => {
             setTimeout(() => {
               reject(new Error('trace should finish before timeout'));
-            }, 1000);
+            }, 2000);
           });
 
           // Read the first chunk to confirm the stream is live