feat(browser): Defer fetch span end until stream body resolves

dev-packages/browser-integration-tests/suites/tracing/request/fetch-streamed-track-stream-performance/init.js

@@ -0,0 +1,14 @@
+import * as Sentry from '@sentry/browser';
+
+window.Sentry = Sentry;
+
+Sentry.init({
+  dsn: 'https://public@dsn.ingest.sentry.io/1337',
+  integrations: [
+    Sentry.browserTracingIntegration({ trackFetchStreamPerformance: true }),
+    Sentry.spanStreamingIntegration(),
+  ],
+  tracePropagationTargets: ['http://sentry-test-site.example'],
+  tracesSampleRate: 1,
+  autoSessionTracking: false,
+});

dev-packages/browser-integration-tests/suites/tracing/request/fetch-streamed-track-stream-performance/subject.js

@@ -0,0 +1 @@
+fetch('http://sentry-test-site.example/delayed').then(res => res.text());

dev-packages/browser-integration-tests/suites/tracing/request/fetch-streamed-track-stream-performance/test.ts

@@ -0,0 +1,43 @@
+import { expect } from '@playwright/test';
+import { sentryTest } from '../../../../utils/fixtures';
+import { shouldSkipTracingTest } from '../../../../utils/helpers';
+import { getSpanOp, waitForStreamedSpans } from '../../../../utils/spanUtils';
+
+sentryTest(
+  'span has correct attributes when trackFetchStreamPerformance is enabled',
+  async ({ getLocalTestUrl, page }) => {
+    sentryTest.skip(shouldSkipTracingTest());
+
+    await page.route('http://sentry-test-site.example/*', route => route.fulfill({ body: 'ok', status: 200 }));
+
+    const url = await getLocalTestUrl({ testDir: __dirname });
+
+    const spansPromise = waitForStreamedSpans(page, spans => spans.some(s => getSpanOp(s) === 'http.client'));
+
+    await page.goto(url);
+
+    const allSpans = await spansPromise;
+    const pageloadSpan = allSpans.find(s => getSpanOp(s) === 'pageload');
+    const requestSpan = allSpans.find(s => getSpanOp(s) === 'http.client');
+
+    expect(requestSpan).toBeDefined();
+    expect(requestSpan!.end_timestamp).toBeGreaterThan(requestSpan!.start_timestamp);
+    expect(requestSpan).toMatchObject({
+      name: 'GET http://sentry-test-site.example/delayed',
+      parent_span_id: pageloadSpan?.span_id,
+      span_id: expect.stringMatching(/[a-f\d]{16}/),
+      start_timestamp: expect.any(Number),
+      end_timestamp: expect.any(Number),
+      trace_id: pageloadSpan?.trace_id,
+      status: 'ok',
+      attributes: expect.objectContaining({
+        'http.method': { type: 'string', value: 'GET' },
+        'http.url': { type: 'string', value: 'http://sentry-test-site.example/delayed' },
+        url: { type: 'string', value: 'http://sentry-test-site.example/delayed' },
+        'server.address': { type: 'string', value: 'sentry-test-site.example' },
+        type: { type: 'string', value: 'fetch' },
+        'http.response.status_code': { type: 'integer', value: 200 },
+      }),
+    });
+  },
+);

packages/browser/src/tracing/request.ts

@@ -1,6 +1,7 @@
 /* eslint-disable max-lines */
 import type {
   Client,
+  HandlerDataFetch,
   HandlerDataXhr,
   RequestHookInfo,
   ResponseHookInfo,
@@ -124,7 +125,11 @@ export interface RequestInstrumentationOptions {
 }
 
 const responseToSpanId = new WeakMap<object, string>();
-const spanIdToEndTimestamp = new Map<string, number>();
+const spanIdToDeferredHandlerData = new Map<string, HandlerDataFetch>();
+const spanIdToFallbackTimeout = new Map<string, ReturnType<typeof setTimeout>>();
+
+// Matches the max fetch timeout defined in core/src/instrument/fetch.ts
+const STREAM_RESOLVE_FALLBACK_MS = 90_000;
 
 export const defaultRequestInstrumentationOptions: RequestInstrumentationOptions = {
   traceFetch: true,
@@ -159,44 +164,66 @@ export function instrumentOutgoingRequests(client: Client, _options?: Partial<Re
   const propagateTraceparent = (client as BrowserClient).getOptions().propagateTraceparent;
 
   if (traceFetch) {
-    // Keeping track of http requests, whose body payloads resolved later than the initial resolved request
-    // e.g. streaming using server sent events (SSE)
-    client.addEventProcessor(event => {
-      if (event.type === 'transaction' && event.spans) {
-        event.spans.forEach(span => {
-          if (span.op === 'http.client') {
-            const updatedTimestamp = spanIdToEndTimestamp.get(span.span_id);
-            if (updatedTimestamp) {
-              span.timestamp = updatedTimestamp / 1000;
-              spanIdToEndTimestamp.delete(span.span_id);
-            }
-          }
-        });
-      }
-      return event;
-    });
-
     if (trackFetchStreamPerformance) {
       addFetchEndInstrumentationHandler(handlerData => {
         if (handlerData.response) {
-          const span = responseToSpanId.get(handlerData.response);
-          if (span && handlerData.endTimestamp) {
-            spanIdToEndTimestamp.set(span, handlerData.endTimestamp);
+          const spanId = responseToSpanId.get(handlerData.response);
+          if (spanId) {
+            const deferredHandlerData = spanIdToDeferredHandlerData.get(spanId);
+            if (deferredHandlerData && handlerData.endTimestamp) {
+              // end span with the correct timestamp
+              deferredHandlerData.endTimestamp = handlerData.endTimestamp;
+              instrumentFetchRequest(deferredHandlerData, shouldCreateSpan, shouldAttachHeadersWithTargets, spans, {
+                propagateTraceparent,
+                onRequestSpanEnd,
+              });
+              spanIdToDeferredHandlerData.delete(spanId);
+
+              // clear fallback timeout since the body was successfully resolved and we ended the span
+              const fallbackTimeout = spanIdToFallbackTimeout.get(spanId);
+              if (fallbackTimeout) {
+                clearTimeout(fallbackTimeout);
+                spanIdToFallbackTimeout.delete(spanId);
+              }
+            }
           }
         }
       });
     }
 
     addFetchInstrumentationHandler(handlerData => {
+      // When tracking streaming performance, defer span end until the response body resolves.
+      // We intercept the end call, save the span, and let the fetchEndInstrumentationHandler
+      // end it with the correct timestamp.
+      if (trackFetchStreamPerformance && handlerData.endTimestamp && handlerData.response) {
+        const spanId = handlerData.fetchData?.__span;
+        if (spanId && spans[spanId]) {
+          responseToSpanId.set(handlerData.response, spanId);
+          spanIdToDeferredHandlerData.set(spanId, handlerData);
+
+          // set fallback timeout to also end the span if the response body is not resolved
+          const fallbackTimeout = setTimeout(() => {
+            const deferredHandlerData = spanIdToDeferredHandlerData.get(spanId);
+            if (deferredHandlerData) {
+              instrumentFetchRequest(deferredHandlerData, shouldCreateSpan, shouldAttachHeadersWithTargets, spans, {
+                propagateTraceparent,
+                onRequestSpanEnd,
+              });
+              spanIdToDeferredHandlerData.delete(spanId);
+              spanIdToFallbackTimeout.delete(spanId);
+            }
+          }, STREAM_RESOLVE_FALLBACK_MS);
+
+          spanIdToFallbackTimeout.set(spanId, fallbackTimeout);
+          return;
+        }
+      }
+
       const createdSpan = instrumentFetchRequest(handlerData, shouldCreateSpan, shouldAttachHeadersWithTargets, spans, {
         propagateTraceparent,
         onRequestSpanEnd,
       });
 
-      if (handlerData.response && handlerData.fetchData.__span) {
-        responseToSpanId.set(handlerData.response, handlerData.fetchData.__span);
-      }
-
       // We cannot use `window.location` in the generic fetch instrumentation,
       // but we need it for reliable `server.address` attribute.
       // so we extend this in here

packages/browser/test/tracing/request.test.ts

@@ -12,11 +12,6 @@ beforeAll(() => {
 });
 
 class MockClient implements Partial<Client> {
-  public addEventProcessor: () => void;
-  constructor() {
-    // Mock addEventProcessor function
-    this.addEventProcessor = vi.fn();
-  }
   // @ts-expect-error not returning options for the test
   public getOptions() {
     return {};