Add retry logic for MS Outlook trigger (#2093)

<!--
Ensure the title clearly reflects what was changed.
Provide a clear and concise description of the changes made. The PR
should only contain the changes related to the issue, and no other
unrelated changes.
-->

Fixes OPS-3295


The error comes from ms teams and cannot be reproduced with the same
data (we don't have an error on the next run). So my guess is to add
retry on the trigger and keep tracking it

Author: Roman Snapko

packages/blocks/microsoft-outlook/src/lib/common/graph-retry.ts

@@ -0,0 +1,84 @@
+import { GraphError } from '@microsoft/microsoft-graph-client';
+import { logger } from '@openops/server-shared';
+
+export type RetryOptions = {
+  maxRetries: number;
+  initialDelayMs: number;
+  shouldRetry: (error: unknown) => boolean;
+  onRetry?: (error: unknown, attempt: number, delayMs: number) => void;
+};
+
+export async function withGraphRetry<T>(
+  fn: () => Promise<T>,
+  options: RetryOptions,
+): Promise<T> {
+  const { maxRetries, initialDelayMs, shouldRetry, onRetry } = options;
+
+  if (maxRetries < 1) {
+    throw new Error(`maxRetries must be at least 1, but got ${maxRetries}`);
+  }
+
+  if (initialDelayMs < 0) {
+    throw new Error(
+      `initialDelayMs must be non-negative, but got ${initialDelayMs}`,
+    );
+  }
+
+  let lastError: unknown;
+
+  for (let i = 0; i < maxRetries; i++) {
+    try {
+      return await fn();
+    } catch (e: unknown) {
+      lastError = e;
+
+      if (!shouldRetry(e) || i === maxRetries - 1) {
+        throw e;
+      }
+
+      const delay = initialDelayMs * Math.pow(2, i);
+      if (onRetry) {
+        onRetry(e, i + 1, delay);
+      }
+
+      await new Promise((resolve) => setTimeout(resolve, delay));
+    }
+  }
+
+  throw lastError;
+}
+
+export const DEFAULT_MAX_RETRIES = 3;
+export const DEFAULT_INITIAL_DELAY_MS = 1000;
+
+export type MicrosoftGraphRetryOptions = {
+  maxRetries?: number;
+  initialDelayMs?: number;
+};
+
+export async function microsoftGraphRetry<T>(
+  fn: () => Promise<T>,
+  options: MicrosoftGraphRetryOptions = {},
+): Promise<T> {
+  const {
+    maxRetries = DEFAULT_MAX_RETRIES,
+    initialDelayMs = DEFAULT_INITIAL_DELAY_MS,
+  } = options;
+
+  return withGraphRetry(fn, {
+    maxRetries,
+    initialDelayMs,
+    shouldRetry: (e) => {
+      const isGraphError = e instanceof GraphError;
+      const statusCode = isGraphError ? e.statusCode : undefined;
+      return statusCode !== undefined && statusCode >= 500;
+    },
+    onRetry: (e, attempt, delay) => {
+      const isGraphError = e instanceof GraphError;
+      const statusCode = isGraphError ? e.statusCode : undefined;
+      logger.warn(
+        `Transient Microsoft Graph error ${statusCode} occurred. Retrying in ${delay}ms... (Attempt ${attempt}/${maxRetries})`,
+      );
+    },
+  });
+}

packages/blocks/microsoft-outlook/src/lib/triggers/new-email.ts

@@ -10,6 +10,7 @@ import {
 import { isEmpty, isString } from '@openops/shared';
 import dayjs from 'dayjs';
 import { microsoftOutlookAuth } from '../common/auth';
+import { microsoftGraphRetry } from '../common/graph-retry';
 import { mailFolderIdDropdown } from '../common/props';
 
 function normalizeString(value: string): string {
@@ -146,15 +147,17 @@ async function fetchMessages(
     );
   }
 
-  let response: PageCollection = await req.get();
+  let response: PageCollection = await microsoftGraphRetry(() => req.get());
   do {
     messages.push(...(response.value as Message[]));
 
     if (!shouldFetchAll || !response['@odata.nextLink']) {
       break;
     }
 
-    response = await client.api(response['@odata.nextLink']).get();
+    response = await microsoftGraphRetry(() =>
+      client.api(response['@odata.nextLink'] as string).get(),
+    );
   } while (response.value.length > 0);
 
   return messages;

packages/blocks/microsoft-outlook/test/graph-retry.test.ts

@@ -0,0 +1,223 @@
+import { GraphError } from '@microsoft/microsoft-graph-client';
+import {
+  microsoftGraphRetry,
+  withGraphRetry,
+} from '../src/lib/common/graph-retry';
+
+describe('withGraphRetry', () => {
+  it('should return the result if fn succeeds on the first attempt', async () => {
+    const fn = jest.fn().mockResolvedValue('success');
+    const result = await withGraphRetry(fn, {
+      maxRetries: 3,
+      initialDelayMs: 1,
+      shouldRetry: () => true,
+    });
+    expect(result).toBe('success');
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should retry and succeed if fn fails initially', async () => {
+    const fn = jest
+      .fn()
+      .mockRejectedValueOnce(new Error('fail'))
+      .mockResolvedValueOnce('success');
+    const onRetry = jest.fn();
+
+    const result = await withGraphRetry(fn, {
+      maxRetries: 3,
+      initialDelayMs: 1,
+      shouldRetry: () => true,
+      onRetry,
+    });
+
+    expect(result).toBe('success');
+    expect(fn).toHaveBeenCalledTimes(2);
+    expect(onRetry).toHaveBeenCalledTimes(1);
+    expect(onRetry).toHaveBeenNthCalledWith(1, expect.any(Error), 1, 1);
+  });
+
+  it('should throw the error if fn fails and shouldRetry returns false', async () => {
+    const error = new Error('critical fail');
+    const fn = jest.fn().mockRejectedValue(error);
+    const shouldRetry = jest.fn().mockReturnValue(false);
+
+    await expect(
+      withGraphRetry(fn, {
+        maxRetries: 3,
+        initialDelayMs: 1,
+        shouldRetry,
+      }),
+    ).rejects.toThrow(error);
+
+    expect(fn).toHaveBeenCalledTimes(1);
+    expect(shouldRetry).toHaveBeenCalledWith(error);
+  });
+
+  it('should throw the last error after reaching max retries', async () => {
+    const error = new Error('persistent fail');
+    const fn = jest.fn().mockRejectedValue(error);
+    const onRetry = jest.fn();
+
+    await expect(
+      withGraphRetry(fn, {
+        maxRetries: 3,
+        initialDelayMs: 1,
+        shouldRetry: () => true,
+        onRetry,
+      }),
+    ).rejects.toThrow(error);
+
+    expect(fn).toHaveBeenCalledTimes(3);
+    expect(onRetry).toHaveBeenCalledTimes(2);
+  });
+
+  it('should use exponential backoff for delays', async () => {
+    jest.useFakeTimers();
+    const fn = jest
+      .fn()
+      .mockRejectedValueOnce(new Error('fail 1'))
+      .mockRejectedValueOnce(new Error('fail 2'))
+      .mockResolvedValueOnce('success');
+    const onRetry = jest.fn();
+
+    const promise = withGraphRetry(fn, {
+      maxRetries: 3,
+      initialDelayMs: 10,
+      shouldRetry: () => true,
+      onRetry,
+    });
+
+    await Promise.resolve();
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(fn).toHaveBeenCalledTimes(1);
+
+    jest.advanceTimersByTime(10);
+    await Promise.resolve();
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(fn).toHaveBeenCalledTimes(2);
+    expect(onRetry).toHaveBeenNthCalledWith(1, expect.any(Error), 1, 10);
+
+    jest.advanceTimersByTime(20);
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(fn).toHaveBeenCalledTimes(3);
+    expect(onRetry).toHaveBeenNthCalledWith(2, expect.any(Error), 2, 20);
+
+    const result = await promise;
+    expect(result).toBe('success');
+
+    jest.useRealTimers();
+  });
+
+  it('should throw error if maxRetries is less than 1', async () => {
+    const fn = jest.fn();
+    await expect(
+      withGraphRetry(fn, {
+        maxRetries: 0,
+        initialDelayMs: 1,
+        shouldRetry: () => true,
+      }),
+    ).rejects.toThrow('maxRetries must be at least 1, but got 0');
+  });
+
+  it('should throw error if initialDelayMs is negative', async () => {
+    const fn = jest.fn();
+    await expect(
+      withGraphRetry(fn, {
+        maxRetries: 1,
+        initialDelayMs: -1,
+        shouldRetry: () => true,
+      }),
+    ).rejects.toThrow('initialDelayMs must be non-negative, but got -1');
+  });
+});
+
+describe('microsoftGraphRetry', () => {
+  it('should retry on 500+ errors', async () => {
+    const error500 = new GraphError();
+    error500.statusCode = 500;
+
+    const error502 = new GraphError();
+    error502.statusCode = 502;
+
+    const fn = jest
+      .fn()
+      .mockRejectedValueOnce(error500)
+      .mockRejectedValueOnce(error502)
+      .mockResolvedValueOnce('success');
+
+    const result = await microsoftGraphRetry(fn, { initialDelayMs: 0 });
+
+    expect(result).toBe('success');
+    expect(fn).toHaveBeenCalledTimes(3);
+  });
+
+  it('should use default values if options are not provided', async () => {
+    const error502 = new GraphError();
+    error502.statusCode = 502;
+
+    const fn = jest
+      .fn()
+      .mockRejectedValueOnce(error502)
+      .mockRejectedValueOnce(error502)
+      .mockResolvedValueOnce('success');
+
+    jest.useFakeTimers();
+    const promise = microsoftGraphRetry(fn);
+
+    await Promise.resolve();
+    await Promise.resolve();
+    await Promise.resolve();
+    jest.advanceTimersByTime(1000);
+    await Promise.resolve();
+    await Promise.resolve();
+    await Promise.resolve();
+    jest.advanceTimersByTime(2000);
+    await Promise.resolve();
+    await Promise.resolve();
+
+    const result = await promise;
+    expect(result).toBe('success');
+    expect(fn).toHaveBeenCalledTimes(3);
+    jest.useRealTimers();
+  });
+
+  it('should allow overriding default values', async () => {
+    const error502 = new GraphError();
+    error502.statusCode = 502;
+
+    const fn = jest.fn().mockRejectedValue(error502);
+
+    await expect(
+      microsoftGraphRetry(fn, {
+        maxRetries: 2,
+        initialDelayMs: 0,
+      }),
+    ).rejects.toThrow(error502);
+
+    expect(fn).toHaveBeenCalledTimes(2);
+  });
+
+  it('should not retry on other errors (e.g., 401)', async () => {
+    const error401 = new GraphError();
+    error401.statusCode = 401;
+
+    const fn = jest.fn().mockRejectedValue(error401);
+
+    await expect(microsoftGraphRetry(fn)).rejects.toThrow(error401);
+
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should not retry on non-GraphError', async () => {
+    const error = new Error('not a graph error');
+
+    const fn = jest.fn().mockRejectedValue(error);
+
+    await expect(microsoftGraphRetry(fn)).rejects.toThrow(error);
+
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+});