feat(core): Ensure replay envelopes are sent in order when offline

Author: timfish

packages/browser/src/transports/offline.ts

@@ -48,8 +48,8 @@ function keys(store: IDBObjectStore): Promise<number[]> {
   return promisifyRequest(store.getAllKeys() as IDBRequest<number[]>);
 }
 
-/** Insert into the store */
-export function insert(store: Store, value: Uint8Array | string, maxQueueSize: number): Promise<void> {
+/** Insert into the end of the store */
+export function push(store: Store, value: Uint8Array | string, maxQueueSize: number): Promise<void> {
   return store(store => {
     return keys(store).then(keys => {
       if (keys.length >= maxQueueSize) {
@@ -63,6 +63,21 @@ export function insert(store: Store, value: Uint8Array | string, maxQueueSize: n
   });
 }
 
+/** Insert into the front of the store */
+export function unshift(store: Store, value: Uint8Array | string, maxQueueSize: number): Promise<void> {
+  return store(store => {
+    return keys(store).then(keys => {
+      if (keys.length >= maxQueueSize) {
+        return;
+      }
+
+      // We insert with an incremented key so that the entries are popped in order
+      store.put(value, Math.min(...keys, 0) - 1);
+      return promisifyRequest(store.transaction);
+    });
+  });
+}
+
 /** Pop the oldest value from the store */
 export function pop(store: Store): Promise<Uint8Array | string | undefined> {
   return store(store => {
@@ -79,7 +94,7 @@ export function pop(store: Store): Promise<Uint8Array | string | undefined> {
   });
 }
 
-export interface BrowserOfflineTransportOptions extends OfflineTransportOptions {
+export interface BrowserOfflineTransportOptions extends Omit<OfflineTransportOptions, 'createStore'> {
   /**
    * Name of indexedDb database to store envelopes in
    * Default: 'sentry-offline'
@@ -110,10 +125,18 @@ function createIndexedDbStore(options: BrowserOfflineTransportOptions): OfflineS
   }
 
   return {
-    insert: async (env: Envelope) => {
+    push: async (env: Envelope) => {
+      try {
+        const serialized = await serializeEnvelope(env);
+        await push(getStore(), serialized, options.maxQueueSize || 30);
+      } catch (_) {
+        //
+      }
+    },
+    unshift: async (env: Envelope) => {
       try {
         const serialized = await serializeEnvelope(env);
-        await insert(getStore(), serialized, options.maxQueueSize || 30);
+        await unshift(getStore(), serialized, options.maxQueueSize || 30);
       } catch (_) {
         //
       }

packages/browser/test/unit/transports/offline.test.ts

@@ -11,7 +11,7 @@ import type {
 import { createEnvelope } from '@sentry/utils';
 
 import { MIN_DELAY } from '../../../../core/src/transports/offline';
-import { createStore, insert, makeBrowserOfflineTransport, pop } from '../../../src/transports/offline';
+import { createStore, makeBrowserOfflineTransport, pop, push, unshift } from '../../../src/transports/offline';
 
 function deleteDatabase(name: string): Promise<void> {
   return new Promise<void>((resolve, reject) => {
@@ -63,21 +63,24 @@ describe('makeOfflineTransport', () => {
     (global as any).TextDecoder = TextDecoder;
   });
 
-  it('indexedDb wrappers insert and pop', async () => {
+  it('indexedDb wrappers push, unshift and pop', async () => {
     const store = createStore('test', 'test');
     const found = await pop(store);
     expect(found).toBeUndefined();
 
-    await insert(store, 'test1', 30);
-    await insert(store, new Uint8Array([1, 2, 3, 4, 5]), 30);
+    await push(store, 'test1', 30);
+    await push(store, new Uint8Array([1, 2, 3, 4, 5]), 30);
+    await unshift(store, 'test2', 30);
 
     const found2 = await pop(store);
-    expect(found2).toEqual('test1');
+    expect(found2).toEqual('test2');
     const found3 = await pop(store);
-    expect(found3).toEqual(new Uint8Array([1, 2, 3, 4, 5]));
-
+    expect(found3).toEqual('test1');
     const found4 = await pop(store);
-    expect(found4).toBeUndefined();
+    expect(found4).toEqual(new Uint8Array([1, 2, 3, 4, 5]));
+
+    const found5 = await pop(store);
+    expect(found5).toBeUndefined();
   });
 
   it('Queues and retries envelope if wrapped transport throws error', async () => {
@@ -104,7 +107,7 @@ describe('makeOfflineTransport', () => {
     const result2 = await transport.send(ERROR_ENVELOPE);
     expect(result2).toEqual({ statusCode: 200 });
 
-    await delay(MIN_DELAY * 2);
+    await delay(MIN_DELAY * 5);
 
     expect(queuedCount).toEqual(1);
     expect(getSendCount()).toEqual(2);

packages/core/src/transports/offline.ts

@@ -12,7 +12,8 @@ function log(msg: string, error?: Error): void {
 }
 
 export interface OfflineStore {
-  insert(env: Envelope): Promise<void>;
+  push(env: Envelope): Promise<void>;
+  unshift(env: Envelope): Promise<void>;
   pop(): Promise<Envelope | undefined>;
 }
 
@@ -55,17 +56,19 @@ export function makeOfflineTransport<TO>(
 ): (options: TO & OfflineTransportOptions) => Transport {
   return options => {
     const transport = createTransport(options);
-    const store = options.createStore ? options.createStore(options) : undefined;
+
+    if (!options.createStore) {
+      throw new Error('No `createStore` function was provided');
+    }
+
+    const store = options.createStore(options);
 
     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.
-      // We also want to drop client reports because they can be generated when we retry sending events while offline.
-      if (envelopeContainsItemType(env, ['replay_event', 'replay_recording', 'client_report'])) {
+      // We want to drop client reports because they can be generated when we retry sending events while offline.
+      if (envelopeContainsItemType(env, ['client_report'])) {
         return false;
       }
 
@@ -77,10 +80,6 @@ export function makeOfflineTransport<TO>(
     }
 
     function flushIn(delay: number): void {
-      if (!store) {
-        return;
-      }
-
       if (flushTimer) {
         clearTimeout(flushTimer as ReturnType<typeof setTimeout>);
       }
@@ -91,7 +90,7 @@ export function makeOfflineTransport<TO>(
         const found = await store.pop();
         if (found) {
           log('Attempting to send previously queued event');
-          void send(found).catch(e => {
+          void send(found, true).catch(e => {
             log('Failed to retry sending', e);
           });
         }
@@ -113,7 +112,15 @@ export function makeOfflineTransport<TO>(
       retryDelay = Math.min(retryDelay * 2, MAX_DELAY);
     }
 
-    async function send(envelope: Envelope): Promise<TransportMakeRequestResponse> {
+    async function send(envelope: Envelope, isRetry: boolean = false): Promise<TransportMakeRequestResponse> {
+      // We queue all replay envelopes to avoid multiple replay envelopes being sent at the same time. If one fails, we
+      // need to retry them in order.
+      if (!isRetry && envelopeContainsItemType(envelope, ['replay_event', 'replay_recording'])) {
+        await store.push(envelope);
+        flushIn(MIN_DELAY);
+        return {};
+      }
+
       try {
         const result = await transport.send(envelope);
 
@@ -133,8 +140,13 @@ export function makeOfflineTransport<TO>(
         retryDelay = START_DELAY;
         return result;
       } catch (e) {
-        if (store && (await shouldQueue(envelope, e as Error, retryDelay))) {
-          await store.insert(envelope);
+        if (await shouldQueue(envelope, e as Error, retryDelay)) {
+          // If this envelope was a retry, we want to add it to the front of the queue so it's retried again first.
+          if (isRetry) {
+            await store.unshift(envelope);
+          } else {
+            await store.push(envelope);
+          }
           flushWithBackOff();
           log('Error sending. Event queued', e as Error);
           return {};