cater for retry-after

Author: timfish

packages/core/src/transports/offline.ts

@@ -1,11 +1,23 @@
 import type { Envelope, InternalBaseTransportOptions, Transport, TransportMakeRequestResponse } from '@sentry/types';
-import { forEachEnvelopeItem, logger } from '@sentry/utils';
+import { forEachEnvelopeItem, logger, parseRetryAfterHeader } from '@sentry/utils';
 
-export const BETWEEN_DELAY = 100; // 100 ms
+export const MIN_DELAY = 100; // 100 ms
 export const START_DELAY = 5_000; // 5 seconds
 const MAX_DELAY = 3.6e6; // 1 hour
 const DEFAULT_QUEUE_SIZE = 30;
 
+function isReplayEnvelope(envelope: Envelope): boolean {
+  let isReplay = false;
+
+  forEachEnvelopeItem(envelope, (_, type) => {
+    if (type === 'replay_event') {
+      isReplay = true;
+    }
+  });
+
+  return isReplay;
+}
+
 type MaybeAsync<T> = T | Promise<T>;
 
 interface OfflineTransportOptions extends InternalBaseTransportOptions {
@@ -44,18 +56,6 @@ export type CreateOfflineStore = (maxQueueCount: number) => OfflineStore;
 
 type Timer = number | { unref?: () => void };
 
-function isReplayEnvelope(envelope: Envelope): boolean {
-  let isReplay = false;
-
-  forEachEnvelopeItem(envelope, (_, type) => {
-    if (type === 'replay_event') {
-      isReplay = true;
-    }
-  });
-
-  return isReplay;
-}
-
 /**
  * Wraps a transport and stores and retries events when they fail to send.
  *
@@ -74,6 +74,10 @@ export function makeOfflineTransport<TO>(
     let retryDelay = START_DELAY;
     let flushTimer: Timer | undefined;
 
+    function log(msg: string, error?: Error): void {
+      __DEBUG_BUILD__ && logger.info(`[Offline]: ${msg}`, error);
+    }
+
     function shouldQueue(env: Envelope, error: Error, retryDelay: number): MaybeAsync<boolean> {
       if (isReplayEnvelope(env)) {
         return false;
@@ -86,25 +90,19 @@ export function makeOfflineTransport<TO>(
       return true;
     }
 
-    function flushLater(overrideDelay?: number): void {
+    function flushIn(delay: number): void {
       if (flushTimer) {
-        if (overrideDelay) {
-          clearTimeout(flushTimer as ReturnType<typeof setTimeout>);
-        } else {
-          return;
-        }
+        clearTimeout(flushTimer as ReturnType<typeof setTimeout>);
       }
 
-      const delay = overrideDelay || retryDelay;
-
       flushTimer = setTimeout(async () => {
         flushTimer = undefined;
 
         const found = await store.pop();
         if (found) {
-          __DEBUG_BUILD__ && logger.info('[Offline]: Attempting to send previously queued event');
+          log('Attempting to send previously queued event');
           void send(found).catch(e => {
-            __DEBUG_BUILD__ && logger.info('[Offline]: Failed to send when retrying', e);
+            log('Failed to retry sending', e);
           });
         }
       }, delay) as Timer;
@@ -113,6 +111,14 @@ export function makeOfflineTransport<TO>(
       if (typeof flushTimer !== 'number' && typeof flushTimer.unref === 'function') {
         flushTimer.unref();
       }
+    }
+
+    function flushWithBackOff(): void {
+      if (flushTimer) {
+        return;
+      }
+
+      flushIn(retryDelay);
 
       retryDelay *= 2;
 
@@ -124,18 +130,27 @@ export function makeOfflineTransport<TO>(
     async function send(envelope: Envelope): Promise<void | TransportMakeRequestResponse> {
       try {
         const result = await transport.send(envelope);
-        // If the status code wasn't a server error, reset retryDelay and flush
-        if (result && (result.statusCode || 500) < 400) {
-          retryDelay = START_DELAY;
-          flushLater(BETWEEN_DELAY);
+
+        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 (await shouldQueue(envelope, e, retryDelay)) {
           await store.insert(envelope);
-          flushLater();
-          __DEBUG_BUILD__ && logger.info('[Offline]: Event queued', e);
+          flushWithBackOff();
+          log('Error sending. Event queued', e);
           return {};
         } else {
           throw e;
@@ -144,7 +159,7 @@ export function makeOfflineTransport<TO>(
     }
 
     if (options.flushAtStartup) {
-      flushLater();
+      flushWithBackOff();
     }
 
     return {

packages/core/test/lib/transports/offline.test.ts

@@ -18,7 +18,7 @@ import { TextEncoder } from 'util';
 
 import { createTransport } from '../../../src';
 import type { CreateOfflineStore } from '../../../src/transports/offline';
-import { BETWEEN_DELAY, makeOfflineTransport, START_DELAY } from '../../../src/transports/offline';
+import { makeOfflineTransport, MIN_DELAY, START_DELAY } from '../../../src/transports/offline';
 
 const ERROR_ENVELOPE = createEnvelope<EventEnvelope>({ event_id: 'aa3ff046696b4bc6b609ce6d28fde9e2', sent_at: '123' }, [
   [{ type: 'event' }, { event_id: 'aa3ff046696b4bc6b609ce6d28fde9e2' }] as EventItem,
@@ -140,7 +140,7 @@ describe('makeOfflineTransport', () => {
     expect(queuedCount).toEqual(0);
     expect(getSendCount()).toEqual(1);
 
-    await delay(BETWEEN_DELAY * 2);
+    await delay(MIN_DELAY * 2);
 
     // After a successful send, the store should be checked
     expect(getCalls()).toEqual(['pop']);
@@ -154,7 +154,7 @@ describe('makeOfflineTransport', () => {
 
     expect(result).toEqual({ statusCode: 200 });
 
-    await delay(BETWEEN_DELAY * 3);
+    await delay(MIN_DELAY * 3);
 
     expect(getSendCount()).toEqual(2);
     // After a successful send from the store, the store should be checked again to ensure it's empty
@@ -179,7 +179,7 @@ describe('makeOfflineTransport', () => {
 
     expect(result).toEqual({});
 
-    await delay(BETWEEN_DELAY * 2);
+    await delay(MIN_DELAY * 2);
 
     expect(getSendCount()).toEqual(0);
     expect(queuedCount).toEqual(1);
@@ -204,7 +204,7 @@ describe('makeOfflineTransport', () => {
 
     expect(result).toEqual({ statusCode: 500 });
 
-    await delay(BETWEEN_DELAY * 2);
+    await delay(MIN_DELAY * 2);
 
     expect(getSendCount()).toEqual(1);
     expect(queuedCount).toEqual(0);
@@ -282,4 +282,43 @@ describe('makeOfflineTransport', () => {
     expect(getSendCount()).toEqual(0);
     expect(getCalls()).toEqual([]);
   });
+
+  it('Follows the Retry-After header', async () => {
+    const { getCalls, store } = createTestStore(ERROR_ENVELOPE);
+    const { getSendCount, baseTransport } = createTestTransport(
+      {
+        statusCode: 429,
+        headers: { 'x-sentry-rate-limits': '', 'retry-after': '3' },
+      },
+      { statusCode: 200 },
+    );
+
+    let queuedCount = 0;
+    const transport = makeOfflineTransport(
+      baseTransport,
+      store,
+    )({
+      ...transportOptions,
+      shouldStore: () => {
+        queuedCount += 1;
+        return true;
+      },
+    });
+    const result = await transport.send(ERROR_ENVELOPE);
+
+    expect(result).toEqual({
+      statusCode: 429,
+      headers: { 'x-sentry-rate-limits': '', 'retry-after': '3' },
+    });
+
+    await delay(MIN_DELAY * 2);
+
+    expect(getSendCount()).toEqual(1);
+
+    await delay(4_000);
+
+    expect(getSendCount()).toEqual(2);
+    expect(queuedCount).toEqual(0);
+    expect(getCalls()).toEqual(['pop', 'pop']);
+  }, 7_000);
 });