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

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 },
  );
  });