Merge pull request #13861 from getsentry/prepare-release/8.33.1

.github/workflows/build.yml

@@ -437,7 +437,7 @@ jobs:
         with:
           node-version-file: 'package.json'
       - name: Set up Deno
-        uses: denoland/setup-deno@v1.4.1
+        uses: denoland/setup-deno@v1.5.1
         with:
           deno-version: v1.38.5
       - name: Restore caches

CHANGELOG.md

@@ -10,6 +10,14 @@
 
 - "You miss 100 percent of the chances you don't take. — Wayne Gretzky" — Michael Scott
 
+## 8.33.1
+
+- fix(core): Update trpc middleware types ([#13859](https://github.com/getsentry/sentry-javascript/pull/13859))
+- fix(fetch): Fix memory leak when handling endless streaming
+  ([#13809](https://github.com/getsentry/sentry-javascript/pull/13809))
+
+Work in this release was contributed by @soapproject. Thank you for your contribution!
+
 ## 8.33.0
 
 ### Important Changes

dev-packages/e2e-tests/test-applications/nextjs-t3/src/server/api/trpc.ts

@@ -70,8 +70,8 @@ export const createCallerFactory = t.createCallerFactory;
  */
 export const createTRPCRouter = t.router;
 
-const sentryMiddleware = Sentry.trpcMiddleware({
-  attachRpcInput: true,
-});
-
-export const publicProcedure = t.procedure.use(async opts => sentryMiddleware(opts));
+export const publicProcedure = t.procedure.use(
+  Sentry.trpcMiddleware({
+    attachRpcInput: true,
+  }),
+);

dev-packages/e2e-tests/test-applications/node-express/src/app.ts

@@ -105,9 +105,7 @@ Sentry.addEventProcessor(event => {
 
 export const t = initTRPC.context<Context>().create();
 
-const sentryMiddleware = Sentry.trpcMiddleware({ attachRpcInput: true });
-
-const procedure = t.procedure.use(async opts => sentryMiddleware(opts));
+const procedure = t.procedure.use(Sentry.trpcMiddleware({ attachRpcInput: true }));
 
 export const appRouter = t.router({
   getSomething: procedure.input(z.string()).query(opts => {

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

@@ -45,7 +45,6 @@ test('Waits for sse streaming when sse has been explicitly aborted', async ({ pa
   await fetchButton.click();
 
   const rootSpan = await transactionPromise;
-  console.log(JSON.stringify(rootSpan, null, 2));
   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;
 
@@ -71,7 +70,7 @@ test('Waits for sse streaming when sse has been explicitly aborted', async ({ pa
   expect(consoleBreadcrumb?.message).toBe('Could not fetch sse AbortError: BodyStreamBuffer was aborted');
 });
 
-test('Aborts when stream takes longer than 5s', async ({ page }) => {
+test('Aborts when stream takes longer than 5s, by not updating the span duration', async ({ page }) => {
   await page.goto('/sse');
 
   const transactionPromise = waitForTransaction('react-router-6', async transactionEvent => {
@@ -102,5 +101,5 @@ test('Aborts when stream takes longer than 5s', async ({ page }) => {
   const resolveBodyDuration = Math.round((httpGet.timestamp as number) - httpGet.start_timestamp);
 
   expect(resolveDuration).toBe(0);
-  expect(resolveBodyDuration).toBe(7);
+  expect(resolveBodyDuration).toBe(0);
 });

packages/core/src/trpc.ts

@@ -33,11 +33,15 @@ function captureIfError(nextResult: unknown): void {
   }
 }
 
+type SentryTrpcMiddleware<T> = T extends Promise<unknown> ? T : Promise<T>;
+
 /**
  * Sentry tRPC middleware that captures errors and creates spans for tRPC procedures.
  */
 export function trpcMiddleware(options: SentryTrpcMiddlewareOptions = {}) {
-  return async function <T>(opts: SentryTrpcMiddlewareArguments<T>): Promise<T> {
+  // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+  // @ts-ignore
+  return async function <T>(opts: SentryTrpcMiddlewareArguments<T>): SentryTrpcMiddleware<T> {
     const { path, type, next, rawInput, getRawInput } = opts;
 
     const client = getClient();
@@ -85,6 +89,6 @@ export function trpcMiddleware(options: SentryTrpcMiddlewareOptions = {}) {
           throw e;
         }
       },
-    );
+    ) as SentryTrpcMiddleware<T>;
   };
 }

packages/utils/src/instrument/fetch.ts

@@ -116,40 +116,57 @@ function instrumentFetch(onFetchResolved?: (response: Response) => void, skipNat
 }
 
 async function resolveResponse(res: Response | undefined, onFinishedResolving: () => void): Promise<void> {
-  if (res && res.body && res.body.getReader) {
-    const responseReader = res.body.getReader();
-
-    // eslint-disable-next-line no-inner-declarations
-    async function consumeChunks({ done }: { done: boolean }): Promise<void> {
-      if (!done) {
-        try {
-          // abort reading if read op takes more than 5s
-          const result = await Promise.race([
-            responseReader.read(),
-            new Promise<{ done: boolean }>(res => {
-              setTimeout(() => {
-                res({ done: true });
-              }, 5000);
-            }),
-          ]);
-          await consumeChunks(result);
-        } catch (error) {
-          // handle error if needed
+  if (res && res.body) {
+    const body = res.body;
+    const responseReader = body.getReader();
+
+    // 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);
+
+        if (done) {
+          onFinishedResolving();
+          readingActive = false;
         }
-      } else {
-        return Promise.resolve();
+      } catch (error) {
+        readingActive = false;
+      } finally {
+        clearTimeout(chunkTimeout);
       }
     }
 
-    return responseReader
-      .read()
-      .then(consumeChunks)
-      .then(onFinishedResolving)
-      .catch(() => undefined);
+    clearTimeout(maxFetchDurationTimeout);
+
+    responseReader.releaseLock();
+    body.cancel().then(null, () => {
+      // noop on error
+    });
   }
 }
 
-async function streamHandler(response: Response): Promise<void> {
+function streamHandler(response: Response): void {
   // clone response for awaiting stream
   let clonedResponseForResolving: Response;
   try {
@@ -158,7 +175,8 @@ async function streamHandler(response: Response): Promise<void> {
     return;
   }
 
-  await resolveResponse(clonedResponseForResolving, () => {
+  // eslint-disable-next-line @typescript-eslint/no-floating-promises
+  resolveResponse(clonedResponseForResolving, () => {
     triggerHandlers('fetch-body-resolved', {
       endTimestamp: timestampInSeconds() * 1000,
       response,