test: Fix e2e test race condition by buffering events

.github/workflows/build.yml

@@ -100,6 +100,7 @@ jobs:
               - 'packages/rollup-utils/**'
               - 'packages/utils/**'
               - 'packages/types/**'
+              - 'dev-packages/test-utils/**'
             browser: &browser
               - *shared
               - 'packages/browser/**'

dev-packages/test-utils/src/event-proxy-server.ts

@@ -1,3 +1,5 @@
+/* eslint-disable max-lines */
+
 import * as fs from 'fs';
 import * as http from 'http';
 import type { AddressInfo } from 'net';
@@ -30,12 +32,22 @@ interface SentryRequestCallbackData {
   sentryResponseStatusCode?: number;
 }
 
+interface EventCallbackListener {
+  (data: string): void;
+}
+
 type OnRequest = (
-  eventCallbackListeners: Set<(data: string) => void>,
+  eventCallbackListeners: Set<EventCallbackListener>,
   proxyRequest: http.IncomingMessage,
   proxyRequestBody: string,
+  eventBuffer: BufferedEvent[],
 ) => Promise<[number, string, Record<string, string> | undefined]>;
 
+interface BufferedEvent {
+  timestamp: number;
+  data: string;
+}
+
 /**
  * Start a generic proxy server.
  * The `onRequest` callback receives the incoming request and the request body,
@@ -51,7 +63,8 @@ export async function startProxyServer(
   },
   onRequest?: OnRequest,
 ): Promise<void> {
-  const eventCallbackListeners: Set<(data: string) => void> = new Set();
+  const eventBuffer: BufferedEvent[] = [];
+  const eventCallbackListeners: Set<EventCallbackListener> = new Set();
 
   const proxyServer = http.createServer((proxyRequest, proxyResponse) => {
     const proxyRequestChunks: Uint8Array[] = [];
@@ -76,15 +89,17 @@ export async function startProxyServer(
 
       const callback: OnRequest =
         onRequest ||
-        (async (eventCallbackListeners, proxyRequest, proxyRequestBody) => {
+        (async (eventCallbackListeners, proxyRequest, proxyRequestBody, eventBuffer) => {
+          eventBuffer.push({ data: proxyRequestBody, timestamp: Date.now() });
+
           eventCallbackListeners.forEach(listener => {
             listener(proxyRequestBody);
           });
 
           return [200, '{}', {}];
         });
 
-      callback(eventCallbackListeners, proxyRequest, proxyRequestBody)
+      callback(eventCallbackListeners, proxyRequest, proxyRequestBody, eventBuffer)
         .then(([statusCode, responseBody, responseHeaders]) => {
           proxyResponse.writeHead(statusCode, responseHeaders);
           proxyResponse.write(responseBody, 'utf-8');
@@ -110,12 +125,24 @@ export async function startProxyServer(
     eventCallbackResponse.statusCode = 200;
     eventCallbackResponse.setHeader('connection', 'keep-alive');
 
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    const searchParams = new URL(eventCallbackRequest.url!, 'http://justsomerandombasesothattheurlisparseable.com/')
+      .searchParams;
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    const listenerTimestamp = Number(searchParams.get('timestamp')!);
+
     const callbackListener = (data: string): void => {
       eventCallbackResponse.write(data.concat('\n'), 'utf8');
     };
 
     eventCallbackListeners.add(callbackListener);
 
+    eventBuffer.forEach(bufferedEvent => {
+      if (bufferedEvent.timestamp > listenerTimestamp) {
+        callbackListener(bufferedEvent.data);
+      }
+    });
+
     eventCallbackRequest.on('close', () => {
       eventCallbackListeners.delete(callbackListener);
     });
@@ -142,7 +169,7 @@ export async function startProxyServer(
  * option to this server (like this `tunnel: http://localhost:${port option}/`).
  */
 export async function startEventProxyServer(options: EventProxyServerOptions): Promise<void> {
-  await startProxyServer(options, async (eventCallbackListeners, proxyRequest, proxyRequestBody) => {
+  await startProxyServer(options, async (eventCallbackListeners, proxyRequest, proxyRequestBody, eventBuffer) => {
     const envelopeHeader: EnvelopeItem[0] = JSON.parse(proxyRequestBody.split('\n')[0] as string);
 
     const shouldForwardEventToSentry = options.forwardToSentry != null ? options.forwardToSentry : true;
@@ -199,8 +226,12 @@ export async function startEventProxyServer(options: EventProxyServerOptions): P
         sentryResponseStatusCode: res.status,
       };
 
+      const dataString = Buffer.from(JSON.stringify(data)).toString('base64');
+
+      eventBuffer.push({ data: dataString, timestamp: Date.now() });
+
       eventCallbackListeners.forEach(listener => {
-        listener(Buffer.from(JSON.stringify(data)).toString('base64'));
+        listener(dataString);
       });
 
       const resHeaders: Record<string, string> = {};
@@ -221,24 +252,28 @@ export async function waitForPlainRequest(
   const eventCallbackServerPort = await retrieveCallbackServerPort(proxyServerName);
 
   return new Promise((resolve, reject) => {
-    const request = http.request(`http://localhost:${eventCallbackServerPort}/`, {}, response => {
-      let eventContents = '';
-
-      response.on('error', err => {
-        reject(err);
-      });
+    const request = http.request(
+      `http://localhost:${eventCallbackServerPort}/?timestamp=${Date.now()}`,
+      {},
+      response => {
+        let eventContents = '';
+
+        response.on('error', err => {
+          reject(err);
+        });
 
-      response.on('data', (chunk: Buffer) => {
-        const chunkString = chunk.toString('utf8');
+        response.on('data', (chunk: Buffer) => {
+          const chunkString = chunk.toString('utf8');
 
-        eventContents = eventContents.concat(chunkString);
+          eventContents = eventContents.concat(chunkString);
 
-        if (callback(eventContents)) {
-          response.destroy();
-          return resolve(eventContents);
-        }
-      });
-    });
+          if (callback(eventContents)) {
+            response.destroy();
+            return resolve(eventContents);
+          }
+        });
+      },
+    );
 
     request.end();
   });
@@ -248,48 +283,53 @@ export async function waitForPlainRequest(
 export async function waitForRequest(
   proxyServerName: string,
   callback: (eventData: SentryRequestCallbackData) => Promise<boolean> | boolean,
+  timestamp: number = Date.now(),
 ): Promise<SentryRequestCallbackData> {
   const eventCallbackServerPort = await retrieveCallbackServerPort(proxyServerName);
 
   return new Promise<SentryRequestCallbackData>((resolve, reject) => {
-    const request = http.request(`http://localhost:${eventCallbackServerPort}/`, {}, response => {
-      let eventContents = '';
-
-      response.on('error', err => {
-        reject(err);
-      });
+    const request = http.request(
+      `http://localhost:${eventCallbackServerPort}/?timestamp=${timestamp}`,
+      {},
+      response => {
+        let eventContents = '';
+
+        response.on('error', err => {
+          reject(err);
+        });
 
-      response.on('data', (chunk: Buffer) => {
-        const chunkString = chunk.toString('utf8');
-        chunkString.split('').forEach(char => {
-          if (char === '\n') {
-            const eventCallbackData: SentryRequestCallbackData = JSON.parse(
-              Buffer.from(eventContents, 'base64').toString('utf8'),
-            );
-            const callbackResult = callback(eventCallbackData);
-            if (typeof callbackResult !== 'boolean') {
-              callbackResult.then(
-                match => {
-                  if (match) {
-                    response.destroy();
-                    resolve(eventCallbackData);
-                  }
-                },
-                err => {
-                  throw err;
-                },
+        response.on('data', (chunk: Buffer) => {
+          const chunkString = chunk.toString('utf8');
+          chunkString.split('').forEach(char => {
+            if (char === '\n') {
+              const eventCallbackData: SentryRequestCallbackData = JSON.parse(
+                Buffer.from(eventContents, 'base64').toString('utf8'),
               );
-            } else if (callbackResult) {
-              response.destroy();
-              resolve(eventCallbackData);
+              const callbackResult = callback(eventCallbackData);
+              if (typeof callbackResult !== 'boolean') {
+                callbackResult.then(
+                  match => {
+                    if (match) {
+                      response.destroy();
+                      resolve(eventCallbackData);
+                    }
+                  },
+                  err => {
+                    throw err;
+                  },
+                );
+              } else if (callbackResult) {
+                response.destroy();
+                resolve(eventCallbackData);
+              }
+              eventContents = '';
+            } else {
+              eventContents = eventContents.concat(char);
             }
-            eventContents = '';
-          } else {
-            eventContents = eventContents.concat(char);
-          }
+          });
         });
-      });
-    });
+      },
+    );
 
     request.end();
   });
@@ -299,18 +339,23 @@ export async function waitForRequest(
 export function waitForEnvelopeItem(
   proxyServerName: string,
   callback: (envelopeItem: EnvelopeItem) => Promise<boolean> | boolean,
+  timestamp: number = Date.now(),
 ): Promise<EnvelopeItem> {
   return new Promise((resolve, reject) => {
-    waitForRequest(proxyServerName, async eventData => {
-      const envelopeItems = eventData.envelope[1];
-      for (const envelopeItem of envelopeItems) {
-        if (await callback(envelopeItem)) {
-          resolve(envelopeItem);
-          return true;
+    waitForRequest(
+      proxyServerName,
+      async eventData => {
+        const envelopeItems = eventData.envelope[1];
+        for (const envelopeItem of envelopeItems) {
+          if (await callback(envelopeItem)) {
+            resolve(envelopeItem);
+            return true;
+          }
         }
-      }
-      return false;
-    }).catch(reject);
+        return false;
+      },
+      timestamp,
+    ).catch(reject);
   });
 }
 
@@ -319,15 +364,20 @@ export function waitForError(
   proxyServerName: string,
   callback: (transactionEvent: Event) => Promise<boolean> | boolean,
 ): Promise<Event> {
+  const timestamp = Date.now();
   return new Promise((resolve, reject) => {
-    waitForEnvelopeItem(proxyServerName, async envelopeItem => {
-      const [envelopeItemHeader, envelopeItemBody] = envelopeItem;
-      if (envelopeItemHeader.type === 'event' && (await callback(envelopeItemBody as Event))) {
-        resolve(envelopeItemBody as Event);
-        return true;
-      }
-      return false;
-    }).catch(reject);
+    waitForEnvelopeItem(
+      proxyServerName,
+      async envelopeItem => {
+        const [envelopeItemHeader, envelopeItemBody] = envelopeItem;
+        if (envelopeItemHeader.type === 'event' && (await callback(envelopeItemBody as Event))) {
+          resolve(envelopeItemBody as Event);
+          return true;
+        }
+        return false;
+      },
+      timestamp,
+    ).catch(reject);
   });
 }
 
@@ -336,15 +386,20 @@ export function waitForTransaction(
   proxyServerName: string,
   callback: (transactionEvent: Event) => Promise<boolean> | boolean,
 ): Promise<Event> {
+  const timestamp = Date.now();
   return new Promise((resolve, reject) => {
-    waitForEnvelopeItem(proxyServerName, async envelopeItem => {
-      const [envelopeItemHeader, envelopeItemBody] = envelopeItem;
-      if (envelopeItemHeader.type === 'transaction' && (await callback(envelopeItemBody as Event))) {
-        resolve(envelopeItemBody as Event);
-        return true;
-      }
-      return false;
-    }).catch(reject);
+    waitForEnvelopeItem(
+      proxyServerName,
+      async envelopeItem => {
+        const [envelopeItemHeader, envelopeItemBody] = envelopeItem;
+        if (envelopeItemHeader.type === 'transaction' && (await callback(envelopeItemBody as Event))) {
+          resolve(envelopeItemBody as Event);
+          return true;
+        }
+        return false;
+      },
+      timestamp,
+    ).catch(reject);
   });
 }