Simplify a bit and add some tests

timfish · timfish · commit 26e5b0ca1453 · 2023-01-19T22:44:29.000Z
diff --git a/packages/core/src/transports/offline.ts b/packages/core/src/transports/offline.ts
@@ -1,26 +1,35 @@
 import type { Envelope, InternalBaseTransportOptions, Transport, TransportMakeRequestResponse } from '@sentry/types';
+import { logger } from '@sentry/utils';
+
+export const START_DELAY = 5_000;
+const MAX_DELAY = 2_000_000_000;
+const DEFAULT_QUEUE_SIZE = 30;
 
 function wasRateLimited(result: TransportMakeRequestResponse): boolean {
-  return !!(result.headers && 'x-sentry-rate-limits' in result.headers);
+  return !!(result.headers && 'x-sentry-rate-limits' in result.headers && result.headers['x-sentry-rate-limits']);
 }
 
 type BeforeSendResponse = 'send' | 'queue' | 'drop';
 
 interface OfflineTransportOptions extends InternalBaseTransportOptions {
   /**
-   * The maximum number of days to keep an event in the queue.
+   * The maximum number of events to keep in the offline queue.
+   *
+   * Defaults: 30
    */
-  maxQueueAgeDays?: number;
+  maxQueueSize?: number;
 
   /**
-   * The maximum number of events to keep in the queue.
+   * Flush the offline queue shortly after startup.
+   *
+   * Defaults: false
    */
-  maxQueueCount?: number;
+  flushAtStartup?: boolean;
 
   /**
-   * Called every time the number of requests in the queue changes.
+   * Called when an event is queued .
    */
-  queuedLengthChanged?: (length: number) => void;
+  eventQueued?: () => void;
 
   /**
    * Called before attempting to send an event to Sentry.
@@ -32,38 +41,42 @@ interface OfflineTransportOptions extends InternalBaseTransportOptions {
   beforeSend?: (request: Envelope) => BeforeSendResponse | Promise<BeforeSendResponse>;
 }
 
-interface OfflineTransportStore {
-  add(env: Envelope): Promise<number>;
-  pop(): Promise<[Envelope | undefined, number]>;
+interface OfflineStore {
+  insert(env: Envelope): Promise<void>;
+  pop(): Promise<Envelope | undefined>;
 }
 
-const START_DELAY = 5_000;
-const MAX_DELAY = 2_000_000_000;
+export type CreateOfflineStore = (maxQueueCount: number) => OfflineStore;
 
-/** */
+/**
+ * Wraps a transport and queues events when envelopes fail to send.
+ *
+ * @param createTransport The transport to wrap.
+ * @param createStore The store implementation to use.
+ */
 export function makeOfflineTransport<TO>(
-  transport: (options: TO) => Transport,
-  store: OfflineTransportStore,
+  createTransport: (options: TO) => Transport,
+  createStore: CreateOfflineStore,
 ): (options: TO & OfflineTransportOptions) => Transport {
   return (options: TO & OfflineTransportOptions) => {
-    const baseTransport = transport(options);
+    const baseTransport = createTransport(options);
+    const maxQueueSize = options.maxQueueSize === undefined ? DEFAULT_QUEUE_SIZE : options.maxQueueSize;
+    const store = createStore(maxQueueSize);
 
     let retryDelay = START_DELAY;
-    let lastQueueLength = -1;
 
-    function queueLengthChanged(length: number): void {
-      if (options.queuedLengthChanged && length !== lastQueueLength) {
-        lastQueueLength = length;
-        options.queuedLengthChanged(length);
+    function queued(): void {
+      if (options.eventQueued) {
+        options.eventQueued();
       }
     }
 
     function queueRequest(envelope: Envelope): Promise<void> {
-      return store.add(envelope).then(count => {
-        queueLengthChanged(count);
+      return store.insert(envelope).then(() => {
+        queued();
 
         setTimeout(() => {
-          flushQueue();
+          void flushQueue();
         }, retryDelay);
 
         retryDelay *= 3;
@@ -76,36 +89,40 @@ export function makeOfflineTransport<TO>(
       });
     }
 
-    function flushQueue(): void {
-      void store.pop().then(([found, count]) => {
-        if (found) {
-          // We have pending plus just found
-          queueLengthChanged(count + 1);
-          void send(found);
-        } else {
-          queueLengthChanged(0);
-        }
-      });
+    async function flushQueue(): Promise<void> {
+      const found = await store.pop();
+
+      if (found) {
+        __DEBUG_BUILD__ && logger.info('[Offline]: Attempting to send previously queued event');
+        void send(found);
+      }
     }
 
-    // eslint-disable-next-line @sentry-internal/sdk/no-async-await
     async function send(request: Envelope): Promise<void | TransportMakeRequestResponse> {
-      let action = (await options.beforeSend?.(request)) || 'send';
+      let action = 'send';
+
+      if (options.beforeSend) {
+        action = await options.beforeSend(request);
+      }
 
       if (action === 'send') {
         try {
           const result = await baseTransport.send(request);
-          if (!wasRateLimited(result || {})) {
+          if (wasRateLimited(result || {})) {
+            __DEBUG_BUILD__ && logger.info('[Offline]: Event queued due to rate limiting');
+            action = 'queue';
+          } else {
+            // Envelope was successfully sent
             // Reset the retry delay
             retryDelay = START_DELAY;
-            // We were successful so check the queue
-            flushQueue();
+            // Check if there are any more in the queue
+            void flushQueue();
             return result;
           }
-        } catch (_) {
-          //
+        } catch (e) {
+          __DEBUG_BUILD__ && logger.info('[Offline]: Event queued due to error', e);
+          action = 'queue';
         }
-        action = 'queue';
       }
 
       if (action == 'queue') {
@@ -115,6 +132,12 @@ export function makeOfflineTransport<TO>(
       return {};
     }
 
+    if (options.flushAtStartup) {
+      setTimeout(() => {
+        void flushQueue();
+      }, retryDelay);
+    }
+
     return {
       send,
       flush: (timeout?: number) => baseTransport.flush(timeout),
diff --git a/packages/core/test/lib/transports/offline.test.ts b/packages/core/test/lib/transports/offline.test.ts
@@ -0,0 +1,175 @@
+import type {
+  EventEnvelope,
+  EventItem,
+  TransportMakeRequestResponse,
+  Envelope,
+  Transport,
+  InternalBaseTransportOptions,
+} from '@sentry/types';
+import { createEnvelope } from '@sentry/utils';
+import { TextEncoder } from 'util';
+
+import { makeOfflineTransport, createTransport } from '../../../src';
+import { CreateOfflineStore, START_DELAY } from '../../../src/transports/offline';
+
+const ERROR_ENVELOPE = createEnvelope<EventEnvelope>({ event_id: 'aa3ff046696b4bc6b609ce6d28fde9e2', sent_at: '123' }, [
+  [{ type: 'event' }, { event_id: 'aa3ff046696b4bc6b609ce6d28fde9e2' }] as EventItem,
+]);
+
+const transportOptions = {
+  recordDroppedEvent: () => undefined, // noop
+  textEncoder: new TextEncoder(),
+};
+
+type MockResult<T> = T | Error;
+
+const createTestTransport = (
+  ...sendResults: MockResult<TransportMakeRequestResponse>[]
+): { getSendCount: () => number; baseTransport: (options: InternalBaseTransportOptions) => Transport } => {
+  let sendCount = 0;
+
+  return {
+    getSendCount: () => sendCount,
+    baseTransport: (options: InternalBaseTransportOptions) =>
+      createTransport(options, () => {
+        return new Promise((resolve, reject) => {
+          const next = sendResults.shift();
+
+          if (next instanceof Error) {
+            reject(next);
+          } else {
+            sendCount += 1;
+            resolve(next as TransportMakeRequestResponse | undefined);
+          }
+        });
+      }),
+  };
+};
+
+type StoreEvents = ('add' | 'pop')[];
+
+function createTestStore(...popResults: MockResult<Envelope | undefined>[]): {
+  getCalls: () => StoreEvents;
+  store: CreateOfflineStore;
+} {
+  const calls: StoreEvents = [];
+
+  return {
+    getCalls: () => calls,
+    store: (maxQueueCount: number) => ({
+      insert: async env => {
+        if (popResults.length < maxQueueCount) {
+          popResults.push(env);
+          calls.push('add');
+        }
+      },
+      pop: async () => {
+        calls.push('pop');
+        const next = popResults.shift();
+
+        if (next instanceof Error) {
+          throw next;
+        }
+
+        return next;
+      },
+    }),
+  };
+}
+
+function delay(ms: number): Promise<void> {
+  return new Promise(resolve => setTimeout(resolve, ms));
+}
+
+describe('makeOfflineTransport', () => {
+  it('Sends envelope and checks the store for further envelopes', async () => {
+    expect.assertions(3);
+
+    const { getCalls, store } = createTestStore();
+    const { getSendCount, baseTransport } = createTestTransport({ statusCode: 200 });
+    const transport = makeOfflineTransport(baseTransport, store)(transportOptions);
+    const result = await transport.send(ERROR_ENVELOPE);
+
+    expect(result).toEqual({ statusCode: 200 });
+    expect(getSendCount()).toEqual(1);
+    // After a successful send, the store should be checked
+    expect(getCalls()).toEqual(['pop']);
+  });
+
+  it('After successfully sending, sends further envelopes found in the store', async () => {
+    const { getCalls, store } = createTestStore(ERROR_ENVELOPE);
+    const { getSendCount, baseTransport } = createTestTransport({ statusCode: 200 }, { statusCode: 200 });
+    const transport = makeOfflineTransport(baseTransport, store)(transportOptions);
+    const result = await transport.send(ERROR_ENVELOPE);
+
+    expect(result).toEqual({ statusCode: 200 });
+
+    await delay(100);
+
+    expect(getSendCount()).toEqual(2);
+    // After a successful send, the store should be checked again to ensure it's empty
+    expect(getCalls()).toEqual(['pop', 'pop']);
+  });
+
+  it('Queues envelope if wrapped transport throws error', async () => {
+    const { getCalls, store } = createTestStore();
+    const { getSendCount, baseTransport } = createTestTransport(new Error());
+    const transport = makeOfflineTransport(baseTransport, store)(transportOptions);
+    const result = await transport.send(ERROR_ENVELOPE);
+
+    expect(result).toEqual({});
+
+    await delay(100);
+
+    expect(getSendCount()).toEqual(0);
+    expect(getCalls()).toEqual(['add']);
+  });
+
+  it('Queues envelope if rate limited', async () => {
+    const { getCalls, store } = createTestStore();
+    const { getSendCount, baseTransport } = createTestTransport({
+      headers: { 'x-sentry-rate-limits': 'something', 'retry-after': null },
+    });
+    const transport = makeOfflineTransport(baseTransport, store)(transportOptions);
+    const result = await transport.send(ERROR_ENVELOPE);
+    expect(result).toEqual({});
+
+    await delay(100);
+
+    expect(getSendCount()).toEqual(1);
+    expect(getCalls()).toEqual(['add']);
+  });
+
+  it(
+    'Retries sending envelope after failure',
+    async () => {
+      const { getCalls, store } = createTestStore();
+      const { getSendCount, baseTransport } = createTestTransport(new Error(), { statusCode: 200 });
+      const transport = makeOfflineTransport(baseTransport, store)(transportOptions);
+      const result = await transport.send(ERROR_ENVELOPE);
+      expect(result).toEqual({});
+      expect(getCalls()).toEqual(['add']);
+
+      await delay(START_DELAY + 1_000);
+
+      expect(getSendCount()).toEqual(1);
+      expect(getCalls()).toEqual(['add', 'pop', 'pop']);
+    },
+    START_DELAY + 2_000,
+  );
+
+  it(
+    'When enabled, sends envelopes found in store shortly after startup',
+    async () => {
+      const { getCalls, store } = createTestStore(ERROR_ENVELOPE, ERROR_ENVELOPE);
+      const { getSendCount, baseTransport } = createTestTransport({ statusCode: 200 }, { statusCode: 200 });
+      const _transport = makeOfflineTransport(baseTransport, store)({ ...transportOptions, flushAtStartup: true });
+
+      await delay(START_DELAY + 1_000);
+
+      expect(getSendCount()).toEqual(2);
+      expect(getCalls()).toEqual(['pop', 'pop', 'pop']);
+    },
+    START_DELAY + 2_000,
+  );
+});
