feat(core): Adapt spans for client-side fetch to streaming responses (#…

…12723)

.size-limit.js

@@ -15,7 +15,7 @@ module.exports = [
     path: 'packages/browser/build/npm/esm/index.js',
     import: createImport('init', 'browserTracingIntegration'),
     gzip: true,
-    limit: '35 KB',
+    limit: '36 KB',
   },
   {
     name: '@sentry/browser (incl. Tracing, Replay)',
@@ -29,7 +29,7 @@ module.exports = [
     path: 'packages/browser/build/npm/esm/index.js',
     import: createImport('init', 'browserTracingIntegration', 'replayIntegration'),
     gzip: true,
-    limit: '65 KB',
+    limit: '66 KB',
     modifyWebpackConfig: function (config) {
       const webpack = require('webpack');
       config.plugins.push(
@@ -107,7 +107,7 @@ module.exports = [
     import: createImport('init', 'ErrorBoundary', 'reactRouterV6BrowserTracingIntegration'),
     ignore: ['react/jsx-runtime'],
     gzip: true,
-    limit: '38 KB',
+    limit: '39 KB',
   },
   // Vue SDK (ESM)
   {

dev-packages/e2e-tests/test-applications/react-router-6/package.json

@@ -6,6 +6,7 @@
     "@sentry/react": "latest || *",
     "@types/react": "18.0.0",
     "@types/react-dom": "18.0.0",
+    "express": "4.19.2",
     "react": "18.2.0",
     "react-dom": "18.2.0",
     "react-router-dom": "^6.4.1",
@@ -14,7 +15,9 @@
   },
   "scripts": {
     "build": "react-scripts build",
-    "start": "serve -s build",
+    "start": "run-p start:client start:server",
+    "start:client": "node server/app.js",
+    "start:server": "serve -s build",
     "test": "playwright test",
     "clean": "npx rimraf node_modules pnpm-lock.yaml",
     "test:build": "pnpm install && npx playwright install && pnpm build",
@@ -43,7 +46,8 @@
   "devDependencies": {
     "@playwright/test": "^1.44.1",
     "@sentry-internal/test-utils": "link:../../../test-utils",
-    "serve": "14.0.1"
+    "serve": "14.0.1",
+    "npm-run-all2": "^6.2.0"
   },
   "volta": {
     "extends": "../../package.json"

dev-packages/e2e-tests/test-applications/react-router-6/server/app.js

@@ -0,0 +1,47 @@
+const express = require('express');
+
+const app = express();
+const PORT = 8080;
+
+const wait = time => {
+  return new Promise(resolve => {
+    setTimeout(() => {
+      resolve();
+    }, time);
+  });
+};
+
+async function sseHandler(request, response, timeout = false) {
+  response.headers = {
+    'Content-Type': 'text/event-stream',
+    Connection: 'keep-alive',
+    'Cache-Control': 'no-cache',
+    'Access-Control-Allow-Origin': '*',
+  };
+
+  response.setHeader('Cache-Control', 'no-cache');
+  response.setHeader('Content-Type', 'text/event-stream');
+  response.setHeader('Access-Control-Allow-Origin', '*');
+  response.setHeader('Connection', 'keep-alive');
+
+  response.flushHeaders();
+
+  await wait(2000);
+
+  for (let index = 0; index < 10; index++) {
+    response.write(`data: ${new Date().toISOString()}\n\n`);
+    if (timeout) {
+      await wait(10000);
+    }
+  }
+
+  response.end();
+}
+
+app.get('/sse', (req, res) => sseHandler(req, res));
+
+app.get('/sse-timeout', (req, res) => sseHandler(req, res, true));
+
+app.listen(PORT, () => {
+  console.log(`SSE service listening at http://localhost:${PORT}`);
+});

dev-packages/e2e-tests/test-applications/react-router-6/src/index.tsx

@@ -11,6 +11,7 @@ import {
   useNavigationType,
 } from 'react-router-dom';
 import Index from './pages/Index';
+import SSE from './pages/SSE';
 import User from './pages/User';
 
 const replay = Sentry.replayIntegration();
@@ -48,6 +49,7 @@ root.render(
     <SentryRoutes>
       <Route path="/" element={<Index />} />
       <Route path="/user/:id" element={<User />} />
+      <Route path="/sse" element={<SSE />} />
     </SentryRoutes>
   </BrowserRouter>,
 );

dev-packages/e2e-tests/test-applications/react-router-6/src/pages/SSE.tsx

@@ -0,0 +1,49 @@
+import * as Sentry from '@sentry/react';
+// biome-ignore lint/nursery/noUnusedImports: Need React import for JSX
+import * as React from 'react';
+
+const fetchSSE = async ({ timeout }: { timeout: boolean }) => {
+  Sentry.startSpanManual({ name: 'sse stream using fetch' }, async span => {
+    const res = await Sentry.startSpan({ name: 'sse fetch call' }, async () => {
+      const endpoint = `http://localhost:8080/${timeout ? 'sse-timeout' : 'sse'}`;
+      return await fetch(endpoint);
+    });
+
+    const stream = res.body;
+    const reader = stream?.getReader();
+
+    const readChunk = async () => {
+      const readRes = await reader?.read();
+      if (readRes?.done) {
+        return;
+      }
+
+      new TextDecoder().decode(readRes?.value);
+
+      await readChunk();
+    };
+
+    try {
+      await readChunk();
+    } catch (error) {
+      console.error('Could not fetch sse', error);
+    }
+
+    span.end();
+  });
+};
+
+const SSE = () => {
+  return (
+    <>
+      <button id="fetch-button" onClick={() => fetchSSE({ timeout: false })}>
+        Fetch SSE
+      </button>
+      <button id="fetch-timeout-button" onClick={() => fetchSSE({ timeout: true })}>
+        Fetch timeout SSE
+      </button>
+    </>
+  );
+};
+
+export default SSE;

dev-packages/e2e-tests/test-applications/react-router-6/tests/sse.test.ts

@@ -0,0 +1,69 @@
+import { expect, test } from '@playwright/test';
+import { waitForTransaction } from '@sentry-internal/test-utils';
+import { SpanJSON } from '@sentry/types';
+
+test('Waits for sse streaming when creating spans', async ({ page }) => {
+  await page.goto('/sse');
+
+  const transactionPromise = waitForTransaction('react-router-6', async transactionEvent => {
+    return !!transactionEvent?.transaction && transactionEvent.contexts?.trace?.op === 'pageload';
+  });
+
+  const fetchButton = page.locator('id=fetch-button');
+  await fetchButton.click();
+
+  const rootSpan = await transactionPromise;
+  const sseFetchCall = rootSpan.spans?.filter(span => span.description === 'sse fetch call')[0] as SpanJSON;
+  const httpGet = rootSpan.spans?.filter(span => span.description === 'GET http://localhost:8080/sse')[0] as SpanJSON;
+
+  expect(sseFetchCall).toBeDefined();
+  expect(httpGet).toBeDefined();
+
+  expect(sseFetchCall?.timestamp).toBeDefined();
+  expect(sseFetchCall?.start_timestamp).toBeDefined();
+  expect(httpGet?.timestamp).toBeDefined();
+  expect(httpGet?.start_timestamp).toBeDefined();
+
+  // http headers get sent instantly from the server
+  const resolveDuration = Math.round((sseFetchCall.timestamp as number) - sseFetchCall.start_timestamp);
+
+  // body streams after 2s
+  const resolveBodyDuration = Math.round((httpGet.timestamp as number) - httpGet.start_timestamp);
+
+  expect(resolveDuration).toBe(0);
+  expect(resolveBodyDuration).toBe(2);
+});
+
+test('Aborts when stream takes longer than 5s', async ({ page }) => {
+  await page.goto('/sse');
+
+  const transactionPromise = waitForTransaction('react-router-6', async transactionEvent => {
+    return !!transactionEvent?.transaction && transactionEvent.contexts?.trace?.op === 'pageload';
+  });
+
+  const fetchButton = page.locator('id=fetch-timeout-button');
+  await fetchButton.click();
+
+  const rootSpan = await transactionPromise;
+  const sseFetchCall = rootSpan.spans?.filter(span => span.description === 'sse fetch call')[0] as SpanJSON;
+  const httpGet = rootSpan.spans?.filter(
+    span => span.description === 'GET http://localhost:8080/sse-timeout',
+  )[0] as SpanJSON;
+
+  expect(sseFetchCall).toBeDefined();
+  expect(httpGet).toBeDefined();
+
+  expect(sseFetchCall?.timestamp).toBeDefined();
+  expect(sseFetchCall?.start_timestamp).toBeDefined();
+  expect(httpGet?.timestamp).toBeDefined();
+  expect(httpGet?.start_timestamp).toBeDefined();
+
+  // http headers get sent instantly from the server
+  const resolveDuration = Math.round((sseFetchCall.timestamp as number) - sseFetchCall.start_timestamp);
+
+  // body streams after 10s but client should abort reading after 5s
+  const resolveBodyDuration = Math.round((httpGet.timestamp as number) - httpGet.start_timestamp);
+
+  expect(resolveDuration).toBe(0);
+  expect(resolveBodyDuration).toBe(7);
+});

packages/browser/src/tracing/browserTracingIntegration.ts

@@ -398,7 +398,7 @@ export const browserTracingIntegration = ((_options: Partial<BrowserTracingOptio
         registerInpInteractionListener();
       }
 
-      instrumentOutgoingRequests({
+      instrumentOutgoingRequests(client, {
         traceFetch,
         traceXHR,
         tracePropagationTargets: client.getOptions().tracePropagationTargets,

packages/browser/src/tracing/request.ts

@@ -23,6 +23,7 @@ import {
 import type { Client, HandlerDataXhr, SentryWrappedXMLHttpRequest, Span } from '@sentry/types';
 import {
   BAGGAGE_HEADER_NAME,
+  addFetchEndInstrumentationHandler,
   addFetchInstrumentationHandler,
   browserPerformanceTimeOrigin,
   dynamicSamplingContextToSentryBaggageHeader,
@@ -93,14 +94,17 @@ export interface RequestInstrumentationOptions {
   shouldCreateSpanForRequest?(this: void, url: string): boolean;
 }
 
+const responseToSpanId = new WeakMap<object, string>();
+const spanIdToEndTimestamp = new Map<string, number>();
+
 export const defaultRequestInstrumentationOptions: RequestInstrumentationOptions = {
   traceFetch: true,
   traceXHR: true,
   enableHTTPTimings: true,
 };
 
 /** Registers span creators for xhr and fetch requests  */
-export function instrumentOutgoingRequests(_options?: Partial<RequestInstrumentationOptions>): void {
+export function instrumentOutgoingRequests(client: Client, _options?: Partial<RequestInstrumentationOptions>): void {
   const { traceFetch, traceXHR, shouldCreateSpanForRequest, enableHTTPTimings, tracePropagationTargets } = {
     traceFetch: defaultRequestInstrumentationOptions.traceFetch,
     traceXHR: defaultRequestInstrumentationOptions.traceXHR,
@@ -115,8 +119,39 @@ export function instrumentOutgoingRequests(_options?: Partial<RequestInstrumenta
   const spans: Record<string, Span> = {};
 
   if (traceFetch) {
+    // Keeping track of http requests, whose body payloads resolved later than the intial 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;
+    });
+
+    addFetchEndInstrumentationHandler(handlerData => {
+      if (handlerData.response) {
+        const span = responseToSpanId.get(handlerData.response);
+        if (span && handlerData.endTimestamp) {
+          spanIdToEndTimestamp.set(span, handlerData.endTimestamp);
+        }
+      }
+    });
+
     addFetchInstrumentationHandler(handlerData => {
       const createdSpan = instrumentFetchRequest(handlerData, shouldCreateSpan, shouldAttachHeadersWithTargets, spans);
+
+      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/unit/tracing/request.test.ts

@@ -1,4 +1,5 @@
 import * as browserUtils from '@sentry-internal/browser-utils';
+import type { Client } from '@sentry/types';
 import * as utils from '@sentry/utils';
 import { WINDOW } from '../../../src/helpers';
 
@@ -10,16 +11,27 @@ beforeAll(() => {
   global.Request = {};
 });
 
+class MockClient implements Partial<Client> {
+  public addEventProcessor: () => void;
+  constructor() {
+    // Mock addEventProcessor function
+    this.addEventProcessor = jest.fn();
+  }
+}
+
 describe('instrumentOutgoingRequests', () => {
+  let client: Client;
+
   beforeEach(() => {
     jest.clearAllMocks();
+    client = new MockClient() as unknown as Client;
   });
 
   it('instruments fetch and xhr requests', () => {
     const addFetchSpy = jest.spyOn(utils, 'addFetchInstrumentationHandler');
     const addXhrSpy = jest.spyOn(browserUtils, 'addXhrInstrumentationHandler');
 
-    instrumentOutgoingRequests();
+    instrumentOutgoingRequests(client);
 
     expect(addFetchSpy).toHaveBeenCalledWith(expect.any(Function));
     expect(addXhrSpy).toHaveBeenCalledWith(expect.any(Function));
@@ -28,15 +40,15 @@ describe('instrumentOutgoingRequests', () => {
   it('does not instrument fetch requests if traceFetch is false', () => {
     const addFetchSpy = jest.spyOn(utils, 'addFetchInstrumentationHandler');
 
-    instrumentOutgoingRequests({ traceFetch: false });
+    instrumentOutgoingRequests(client, { traceFetch: false });
 
     expect(addFetchSpy).not.toHaveBeenCalled();
   });
 
   it('does not instrument xhr requests if traceXHR is false', () => {
     const addXhrSpy = jest.spyOn(browserUtils, 'addXhrInstrumentationHandler');
 
-    instrumentOutgoingRequests({ traceXHR: false });
+    instrumentOutgoingRequests(client, { traceXHR: false });
 
     expect(addXhrSpy).not.toHaveBeenCalled();
   });

packages/sveltekit/test/client/browserTracingIntegration.test.ts

@@ -60,7 +60,7 @@ describe('browserTracingIntegration', () => {
       return createdRootSpan as Span;
     });
 
-  const fakeClient = { getOptions: () => ({}), on: () => {} };
+  const fakeClient = { getOptions: () => ({}), on: () => {}, addEventProcessor: () => {} };
 
   const mockedRoutingSpan = {
     end: () => {},