fix(nextjs): Don't share I/O resources in between requests (#6980)

Author: lforst

packages/nextjs/src/edge/transport.ts

@@ -1,5 +1,6 @@
 import { createTransport } from '@sentry/core';
 import type { BaseTransportOptions, Transport, TransportMakeRequestResponse, TransportRequest } from '@sentry/types';
+import { SentryError } from '@sentry/utils';
 
 export interface EdgeTransportOptions extends BaseTransportOptions {
   /** Fetch API init parameters. Used by the FetchTransport */
@@ -8,6 +9,66 @@ export interface EdgeTransportOptions extends BaseTransportOptions {
   headers?: { [key: string]: string };
 }
 
+const DEFAULT_TRANSPORT_BUFFER_SIZE = 30;
+
+/**
+ * This is a modified promise buffer that collects tasks until drain is called.
+ * We need this in the edge runtime because edge function invocations may not share I/O objects, like fetch requests
+ * and responses, and the normal PromiseBuffer inherently buffers stuff inbetween incoming requests.
+ *
+ * A limitation we need to be aware of is that DEFAULT_TRANSPORT_BUFFER_SIZE is the maximum amount of payloads the
+ * SDK can send for a given edge function invocation.
+ */
+export class IsolatedPromiseBuffer {
+  // We just have this field because the promise buffer interface requires it.
+  // If we ever remove it from the interface we should also remove it here.
+  public $: Array<PromiseLike<TransportMakeRequestResponse>> = [];
+
+  private _taskProducers: (() => PromiseLike<TransportMakeRequestResponse>)[] = [];
+
+  public constructor(private readonly _bufferSize: number = DEFAULT_TRANSPORT_BUFFER_SIZE) {}
+
+  /**
+   * @inheritdoc
+   */
+  public add(taskProducer: () => PromiseLike<TransportMakeRequestResponse>): PromiseLike<void> {
+    if (this._taskProducers.length >= this._bufferSize) {
+      return Promise.reject(new SentryError('Not adding Promise because buffer limit was reached.'));
+    }
+
+    this._taskProducers.push(taskProducer);
+    return Promise.resolve();
+  }
+
+  /**
+   * @inheritdoc
+   */
+  public drain(timeout?: number): PromiseLike<boolean> {
+    const oldTaskProducers = [...this._taskProducers];
+    this._taskProducers = [];
+
+    return new Promise(resolve => {
+      const timer = setTimeout(() => {
+        if (timeout && timeout > 0) {
+          resolve(false);
+        }
+      }, timeout);
+
+      void Promise.all(
+        oldTaskProducers.map(taskProducer =>
+          taskProducer().then(null, () => {
+            // catch all failed requests
+          }),
+        ),
+      ).then(() => {
+        // resolve to true if all fetch requests settled
+        clearTimeout(timer);
+        resolve(true);
+      });
+    });
+  }
+}
+
 /**
  * Creates a Transport that uses the Edge Runtimes native fetch API to send events to Sentry.
  */
@@ -21,18 +82,16 @@ export function makeEdgeTransport(options: EdgeTransportOptions): Transport {
       ...options.fetchOptions,
     };
 
-    try {
-      return fetch(options.url, requestOptions).then(response => ({
+    return fetch(options.url, requestOptions).then(response => {
+      return {
         statusCode: response.status,
         headers: {
           'x-sentry-rate-limits': response.headers.get('X-Sentry-Rate-Limits'),
           'retry-after': response.headers.get('Retry-After'),
         },
-      }));
-    } catch (e) {
-      return Promise.reject(e);
-    }
+      };
+    });
   }
 
-  return createTransport(options, makeRequest);
+  return createTransport(options, makeRequest, new IsolatedPromiseBuffer(options.bufferSize));
 }

packages/nextjs/src/edge/utils/edgeWrapperUtils.ts

@@ -44,19 +44,15 @@ export function withEdgeWrapping<H extends EdgeRouteHandler>(
 
         const dynamicSamplingContext = baggageHeaderToDynamicSamplingContext(req.headers.get('baggage'));
 
-        span = startTransaction(
-          {
-            name: options.spanDescription,
-            op: options.spanOp,
-            ...traceparentData,
-            metadata: {
-              dynamicSamplingContext: traceparentData && !dynamicSamplingContext ? {} : dynamicSamplingContext,
-              source: 'route',
-            },
+        span = startTransaction({
+          name: options.spanDescription,
+          op: options.spanOp,
+          ...traceparentData,
+          metadata: {
+            dynamicSamplingContext: traceparentData && !dynamicSamplingContext ? {} : dynamicSamplingContext,
+            source: 'route',
           },
-          // extra context passed to the `tracesSampler`
-          { request: req },
-        );
+        });
       }
 
       currentScope?.setSpan(span);

packages/nextjs/test/edge/edgeWrapperUtils.test.ts

@@ -84,7 +84,6 @@ describe('withEdgeWrapping', () => {
     expect(startTransactionSpy).toHaveBeenCalledTimes(1);
     expect(startTransactionSpy).toHaveBeenCalledWith(
       expect.objectContaining({ metadata: { source: 'route' }, name: 'some label', op: 'some op' }),
-      { request },
     );
   });
 

packages/nextjs/test/edge/transport.test.ts

@@ -3,7 +3,7 @@ import { createEnvelope, serializeEnvelope } from '@sentry/utils';
 import { TextEncoder } from 'util';
 
 import type { EdgeTransportOptions } from '../../src/edge/transport';
-import { makeEdgeTransport } from '../../src/edge/transport';
+import { IsolatedPromiseBuffer, makeEdgeTransport } from '../../src/edge/transport';
 
 const DEFAULT_EDGE_TRANSPORT_OPTIONS: EdgeTransportOptions = {
   url: 'https://sentry.io/api/42/store/?sentry_key=123&sentry_version=7',
@@ -51,6 +51,7 @@ describe('Edge Transport', () => {
 
     expect(mockFetch).toHaveBeenCalledTimes(0);
     await transport.send(ERROR_ENVELOPE);
+    await transport.flush();
     expect(mockFetch).toHaveBeenCalledTimes(1);
 
     expect(mockFetch).toHaveBeenLastCalledWith(DEFAULT_EDGE_TRANSPORT_OPTIONS.url, {
@@ -77,6 +78,7 @@ describe('Edge Transport', () => {
 
     expect(headers.get).toHaveBeenCalledTimes(0);
     await transport.send(ERROR_ENVELOPE);
+    await transport.flush();
 
     expect(headers.get).toHaveBeenCalledTimes(2);
     expect(headers.get).toHaveBeenCalledWith('X-Sentry-Rate-Limits');
@@ -101,10 +103,61 @@ describe('Edge Transport', () => {
     const transport = makeEdgeTransport({ ...DEFAULT_EDGE_TRANSPORT_OPTIONS, fetchOptions: REQUEST_OPTIONS });
 
     await transport.send(ERROR_ENVELOPE);
+    await transport.flush();
     expect(mockFetch).toHaveBeenLastCalledWith(DEFAULT_EDGE_TRANSPORT_OPTIONS.url, {
       body: serializeEnvelope(ERROR_ENVELOPE, new TextEncoder()),
       method: 'POST',
       ...REQUEST_OPTIONS,
     });
   });
 });
+
+describe('IsolatedPromiseBuffer', () => {
+  it('should not call tasks until drained', async () => {
+    const ipb = new IsolatedPromiseBuffer();
+
+    const task1 = jest.fn(() => Promise.resolve({}));
+    const task2 = jest.fn(() => Promise.resolve({}));
+
+    await ipb.add(task1);
+    await ipb.add(task2);
+
+    expect(task1).not.toHaveBeenCalled();
+    expect(task2).not.toHaveBeenCalled();
+
+    await ipb.drain();
+
+    expect(task1).toHaveBeenCalled();
+    expect(task2).toHaveBeenCalled();
+  });
+
+  it('should not allow adding more items than the specified limit', async () => {
+    const ipb = new IsolatedPromiseBuffer(3);
+
+    const task1 = jest.fn(() => Promise.resolve({}));
+    const task2 = jest.fn(() => Promise.resolve({}));
+    const task3 = jest.fn(() => Promise.resolve({}));
+    const task4 = jest.fn(() => Promise.resolve({}));
+
+    await ipb.add(task1);
+    await ipb.add(task2);
+    await ipb.add(task3);
+
+    await expect(ipb.add(task4)).rejects.toThrowError('Not adding Promise because buffer limit was reached.');
+  });
+
+  it('should not throw when one of the tasks throws when drained', async () => {
+    const ipb = new IsolatedPromiseBuffer();
+
+    const task1 = jest.fn(() => Promise.resolve({}));
+    const task2 = jest.fn(() => Promise.reject(new Error()));
+
+    await ipb.add(task1);
+    await ipb.add(task2);
+
+    await expect(ipb.drain()).resolves.toEqual(true);
+
+    expect(task1).toHaveBeenCalled();
+    expect(task2).toHaveBeenCalled();
+  });
+});

packages/nextjs/test/edge/withSentryAPI.test.ts

@@ -53,7 +53,6 @@ describe('wrapApiHandlerWithSentry', () => {
         name: 'POST /user/[userId]/post/[postId]',
         op: 'http.server',
       }),
-      { request },
     );
   });