feat: Use makeOfflineTransport logic for offline support (v5) (#874)

.github/workflows/build.yml

@@ -63,6 +63,7 @@ jobs:
           node-version-file: 'package.json'
       - run: yarn install
       - name: Run Unit Tests
+        timeout-minutes: 10
         run: yarn test
 
   e2e:

package.json

@@ -53,7 +53,7 @@
     "pretest": "yarn build",
     "test": "vitest run --root=./test/unit",
     "pree2e": "rimraf --glob test/e2e/dist/**/node_modules/@sentry/** test/e2e/dist/**/yarn.lock test/e2e/dist/**/package-lock.json && node scripts/clean-cache.js && yarn build && npm pack",
-    "e2e": "xvfb-maybe vitest run --root=./test/e2e"
+    "e2e": "xvfb-maybe vitest run --root=./test/e2e --silent=false --disable-console-intercept"
   },
   "dependencies": {
     "@sentry/browser": "8.0.0-alpha.9",

src/main/transports/electron-offline-net.ts

@@ -1,153 +1,23 @@
-import { createTransport } from '@sentry/core';
-import { Transport, TransportMakeRequestResponse, TransportRequest } from '@sentry/types';
-import { logger } from '@sentry/utils';
-import { net } from 'electron';
-import { join } from 'path';
+import { makeOfflineTransport, OfflineTransportOptions } from '@sentry/core';
+import { BaseTransportOptions, Transport } from '@sentry/types';
 
-import { getSentryCachePath } from '../electron-normalize';
-import { createElectronNetRequestExecutor, ElectronNetTransportOptions } from './electron-net';
-import { PersistedRequestQueue } from './queue';
+import { ElectronNetTransportOptions, makeElectronTransport } from './electron-net';
+import { createOfflineStore, OfflineStoreOptions } from './offline-store';
 
-type BeforeSendResponse = 'send' | 'queue' | 'drop';
-
-export interface ElectronOfflineTransportOptions extends ElectronNetTransportOptions {
-  /**
-   * The maximum number of days to keep an event in the queue.
-   */
-  maxQueueAgeDays?: number;
-
-  /**
-   * The maximum number of events to keep in the queue.
-   */
-  maxQueueCount?: number;
-
-  /**
-   * Called every time the number of requests in the queue changes.
-   */
-  queuedLengthChanged?: (length: number) => void;
-
-  /**
-   * Called before attempting to send an event to Sentry.
-   *
-   * Return 'send' to attempt to send the event.
-   * Return 'queue' to queue and persist the event for sending later.
-   * Return 'drop' to drop the event.
-   */
-  beforeSend?: (request: TransportRequest) => BeforeSendResponse | Promise<BeforeSendResponse>;
-}
-
-const START_DELAY = 5_000;
-const MAX_DELAY = 2_000_000_000;
-
-/** Returns true is there's a chance we're online */
-function maybeOnline(): boolean {
-  return !('online' in net) || net.online === true;
-}
-
-function defaultBeforeSend(_: TransportRequest): BeforeSendResponse {
-  return maybeOnline() ? 'send' : 'queue';
-}
-
-function isRateLimited(result: TransportMakeRequestResponse): boolean {
-  return !!(result.headers && 'x-sentry-rate-limits' in result.headers);
-}
+export type ElectronOfflineTransportOptions = ElectronNetTransportOptions &
+  OfflineTransportOptions &
+  Partial<OfflineStoreOptions>;
 
 /**
  * Creates a Transport that uses Electrons net module to send events to Sentry. When they fail to send they are
  * persisted to disk and sent later
  */
-export function makeElectronOfflineTransport(options: ElectronOfflineTransportOptions): Transport {
-  const netMakeRequest = createElectronNetRequestExecutor(options.url, options.headers || {});
-  const queue: PersistedRequestQueue = new PersistedRequestQueue(
-    join(getSentryCachePath(), 'queue'),
-    options.maxQueueAgeDays,
-    options.maxQueueCount,
-  );
-
-  const beforeSend = options.beforeSend || defaultBeforeSend;
-
-  let retryDelay: number = START_DELAY;
-  let lastQueueLength = -1;
-
-  function queueLengthChanged(queuedEvents: number): void {
-    if (options.queuedLengthChanged && queuedEvents !== lastQueueLength) {
-      lastQueueLength = queuedEvents;
-      options.queuedLengthChanged(queuedEvents);
-    }
-  }
-
-  function flushQueue(): void {
-    queue
-      .pop()
-      .then((found) => {
-        if (found) {
-          // We have pendingCount plus found.request
-          queueLengthChanged(found.pendingCount + 1);
-          logger.log('Found a request in the queue');
-          makeRequest(found.request).catch((e) => logger.error(e));
-        } else {
-          queueLengthChanged(0);
-        }
-      })
-      .catch((e) => logger.error(e));
-  }
-
-  async function queueRequest(request: TransportRequest): Promise<TransportMakeRequestResponse> {
-    logger.log('Queuing request');
-    queueLengthChanged(await queue.add(request));
-
-    setTimeout(() => {
-      flushQueue();
-    }, retryDelay);
-
-    retryDelay *= 3;
-
-    // If the delay is bigger than 2^31 (max signed 32-bit int), setTimeout throws
-    // an error and falls back to 1 which can cause a huge number of requests.
-    if (retryDelay > MAX_DELAY) {
-      retryDelay = MAX_DELAY;
-    }
-
-    return {};
-  }
-
-  async function makeRequest(request: TransportRequest): Promise<TransportMakeRequestResponse> {
-    let action = beforeSend(request);
-
-    if (action instanceof Promise) {
-      action = await action;
-    }
-
-    if (action === 'send') {
-      try {
-        const result = await netMakeRequest(request);
-
-        if (!isRateLimited(result)) {
-          logger.log('Successfully sent');
-          // Reset the retry delay
-          retryDelay = START_DELAY;
-          // We were successful so check the queue
-          flushQueue();
-          return result;
-        } else {
-          logger.log('Rate limited', result.headers);
-        }
-      } catch (error) {
-        logger.log('Error sending:', error);
-      }
-
-      action = 'queue';
-    }
-
-    if (action == 'queue') {
-      return queueRequest(request);
-    }
-
-    logger.log('Dropping request');
-    return {};
-  }
-
-  flushQueue();
-
-  return createTransport(options, makeRequest);
-}
+export const makeElectronOfflineTransport = <T extends BaseTransportOptions>(
+  options: T & ElectronOfflineTransportOptions,
+): Transport => {
+  return makeOfflineTransport(makeElectronTransport)({
+    flushAtStartup: true,
+    ...options,
+    createStore: createOfflineStore,
+  });
+};

src/main/transports/offline-store.ts

@@ -0,0 +1,124 @@
+import { OfflineStore } from '@sentry/core';
+import { Envelope } from '@sentry/types';
+import { logger, parseEnvelope, serializeEnvelope, uuid4 } from '@sentry/utils';
+import { promises as fs } from 'fs';
+import { join } from 'path';
+
+import { getSentryCachePath } from '../electron-normalize';
+import { Store } from '../store';
+
+/** Internal type used to expose the envelope date without having to read it into memory */
+interface PersistedRequest {
+  id: string;
+  date: Date;
+}
+
+export interface OfflineStoreOptions {
+  /**
+   * Path to the offline queue directory.
+   */
+  queuePath: string;
+  /**
+   * Maximum number of days to store requests.
+   */
+  maxAgeDays: number;
+  /**
+   * Maximum number of requests to store.
+   */
+  maxQueueSize: number;
+}
+
+const MILLISECONDS_PER_DAY = 86_400_000;
+
+function isOutdated(request: PersistedRequest, maxAgeDays: number): boolean {
+  const cutOff = Date.now() - MILLISECONDS_PER_DAY * maxAgeDays;
+  return request.date.getTime() < cutOff;
+}
+
+function getSentAtFromEnvelope(envelope: Envelope): Date | undefined {
+  const header = envelope[0];
+  if (typeof header.sent_at === 'string') {
+    return new Date(header.sent_at);
+  }
+  return undefined;
+}
+
+/**
+ * Creates a new offline store.
+ */
+export function createOfflineStore(userOptions: Partial<OfflineStoreOptions>): OfflineStore {
+  function log(...args: unknown[]): void {
+    logger.log(`[Offline Store]:`, ...args);
+  }
+
+  const options: OfflineStoreOptions = {
+    maxAgeDays: userOptions.maxAgeDays || 30,
+    maxQueueSize: userOptions.maxQueueSize || 30,
+    queuePath: userOptions.queuePath || join(getSentryCachePath(), 'queue'),
+  };
+  const queue = new Store<PersistedRequest[]>(options.queuePath, 'queue-v2', []);
+
+  function removeBody(id: string): void {
+    fs.unlink(join(options.queuePath, id)).catch(() => {
+      // ignore
+    });
+  }
+
+  function removeStaleRequests(queue: PersistedRequest[]): void {
+    while (queue[0] && isOutdated(queue[0], options.maxAgeDays)) {
+      const removed = queue.shift() as PersistedRequest;
+      log('Removing stale envelope', removed);
+      removeBody(removed.id);
+    }
+  }
+
+  return {
+    insert: async (env) => {
+      log('Adding envelope to offline storage');
+
+      const id = uuid4();
+
+      try {
+        const data = serializeEnvelope(env);
+        await fs.mkdir(options.queuePath, { recursive: true });
+        await fs.writeFile(join(options.queuePath, id), data);
+      } catch (e) {
+        log('Failed to save', e);
+      }
+
+      await queue.update((queue) => {
+        removeStaleRequests(queue);
+
+        if (queue.length >= options.maxQueueSize) {
+          removeBody(id);
+          return queue;
+        }
+
+        queue.push({ id, date: getSentAtFromEnvelope(env) || new Date() });
+
+        return queue;
+      });
+    },
+    pop: async () => {
+      log('Popping envelope from offline storage');
+      let request: PersistedRequest | undefined;
+      await queue.update((queue) => {
+        removeStaleRequests(queue);
+        request = queue.shift();
+        return queue;
+      });
+
+      if (request) {
+        try {
+          const data = await fs.readFile(join(options.queuePath, request.id));
+          removeBody(request.id);
+          return parseEnvelope(data);
+        } catch (e) {
+          log('Failed to read', e);
+        }
+      }
+
+      return undefined;
+    },
+  };
+}