feat(core): Add Offline Transport wrapper (#6884)

Author: timfish

packages/core/src/index.ts

@@ -1,5 +1,6 @@
 export type { ClientClass } from './sdk';
 export type { Carrier, Layer } from './hub';
+export type { OfflineStore, OfflineTransportOptions } from './transports/offline';
 
 export {
   addBreadcrumb,
@@ -24,6 +25,7 @@ export { getEnvelopeEndpointWithUrlEncodedAuth, getReportDialogEndpoint } from '
 export { BaseClient } from './baseclient';
 export { initAndBind } from './sdk';
 export { createTransport } from './transports/base';
+export { makeOfflineTransport } from './transports/offline';
 export { SDK_VERSION } from './version';
 export { getIntegrationsToSetup } from './integration';
 export { FunctionToString, InboundFilters } from './integrations';

packages/core/src/transports/offline.ts

@@ -0,0 +1,165 @@
+import type { Envelope, InternalBaseTransportOptions, Transport, TransportMakeRequestResponse } from '@sentry/types';
+import { forEachEnvelopeItem, logger, parseRetryAfterHeader } from '@sentry/utils';
+
+export const MIN_DELAY = 100; // 100 ms
+export const START_DELAY = 5_000; // 5 seconds
+const MAX_DELAY = 3.6e6; // 1 hour
+
+function isReplayEnvelope(envelope: Envelope): boolean {
+  let isReplay = false;
+
+  forEachEnvelopeItem(envelope, (_, type) => {
+    if (type === 'replay_event') {
+      isReplay = true;
+    }
+  });
+
+  return isReplay;
+}
+
+function log(msg: string, error?: Error): void {
+  __DEBUG_BUILD__ && logger.info(`[Offline]: ${msg}`, error);
+}
+
+export interface OfflineStore {
+  insert(env: Envelope): Promise<void>;
+  pop(): Promise<Envelope | undefined>;
+}
+
+export type CreateOfflineStore = (options: OfflineTransportOptions) => OfflineStore;
+
+export interface OfflineTransportOptions extends InternalBaseTransportOptions {
+  /**
+   * A function that creates the offline store instance.
+   */
+  createStore?: CreateOfflineStore;
+
+  /**
+   * Flush the offline store shortly after startup.
+   *
+   * Defaults: false
+   */
+  flushAtStartup?: boolean;
+
+  /**
+   * Called before an event is stored.
+   *
+   * Return false to drop the envelope rather than store it.
+   *
+   * @param envelope The envelope that failed to send.
+   * @param error The error that occurred.
+   * @param retryDelay The current retry delay in milliseconds.
+   */
+  shouldStore?: (envelope: Envelope, error: Error, retryDelay: number) => boolean | Promise<boolean>;
+}
+
+type Timer = number | { unref?: () => void };
+
+/**
+ * Wraps a transport and stores and retries events when they fail to send.
+ *
+ * @param createTransport The transport to wrap.
+ */
+export function makeOfflineTransport<TO>(
+  createTransport: (options: TO) => Transport,
+): (options: TO & OfflineTransportOptions) => Transport {
+  return options => {
+    const transport = createTransport(options);
+    const store = options.createStore ? options.createStore(options) : undefined;
+
+    let retryDelay = START_DELAY;
+    let flushTimer: Timer | undefined;
+
+    function shouldQueue(env: Envelope, error: Error, retryDelay: number): boolean | Promise<boolean> {
+      // We don't queue Session Replay envelopes because they are:
+      // - Ordered and Replay relies on the response status to know when they're successfully sent.
+      // - Likely to fill the queue quickly and block other events from being sent.
+      if (isReplayEnvelope(env)) {
+        return false;
+      }
+
+      if (options.shouldStore) {
+        return options.shouldStore(env, error, retryDelay);
+      }
+
+      return true;
+    }
+
+    function flushIn(delay: number): void {
+      if (!store) {
+        return;
+      }
+
+      if (flushTimer) {
+        clearTimeout(flushTimer as ReturnType<typeof setTimeout>);
+      }
+
+      flushTimer = setTimeout(async () => {
+        flushTimer = undefined;
+
+        const found = await store.pop();
+        if (found) {
+          log('Attempting to send previously queued event');
+          void send(found).catch(e => {
+            log('Failed to retry sending', e);
+          });
+        }
+      }, delay) as Timer;
+
+      // We need to unref the timer in node.js, otherwise the node process never exit.
+      if (typeof flushTimer !== 'number' && flushTimer.unref) {
+        flushTimer.unref();
+      }
+    }
+
+    function flushWithBackOff(): void {
+      if (flushTimer) {
+        return;
+      }
+
+      flushIn(retryDelay);
+
+      retryDelay = Math.min(retryDelay * 2, MAX_DELAY);
+    }
+
+    async function send(envelope: Envelope): Promise<void | TransportMakeRequestResponse> {
+      try {
+        const result = await transport.send(envelope);
+
+        let delay = MIN_DELAY;
+
+        if (result) {
+          // If there's a retry-after header, use that as the next delay.
+          if (result.headers && result.headers['retry-after']) {
+            delay = parseRetryAfterHeader(result.headers['retry-after']);
+          } // If we have a server error, return now so we don't flush the queue.
+          else if ((result.statusCode || 0) >= 400) {
+            return result;
+          }
+        }
+
+        flushIn(delay);
+        retryDelay = START_DELAY;
+        return result;
+      } catch (e) {
+        if (store && (await shouldQueue(envelope, e, retryDelay))) {
+          await store.insert(envelope);
+          flushWithBackOff();
+          log('Error sending. Event queued', e);
+          return {};
+        } else {
+          throw e;
+        }
+      }
+    }
+
+    if (options.flushAtStartup) {
+      flushWithBackOff();
+    }
+
+    return {
+      send,
+      flush: t => transport.flush(t),
+    };
+  };
+}