Respect Gmail retry-after in messaging throttle (#17850)

Gmail 429/403 rate-limit responses include an explicit retry-after timestamp, usually ~15 minutes out. The exponential backoff starts at 1 minute, so the channel burns through all 5 retry attempts before the window actually closes and gets marked as permanently failed. Adds throttleRetryAfter to the message channel and uses max(backoff, retryAfter) in isThrottled().
2026-02-13 22:50:37 +05:30
parent 84afbb4d2c
commit d54b713264
21 changed files with 1235 additions and 866 deletions
@@ -116,6 +116,7 @@ export class MessageChannelSyncStatusService {
        syncCursor: '',
        syncStageStartedAt: null,
        throttleFailureCount: 0,
+        throttleRetryAfter: null,
        pendingGroupEmailsAction: MessageChannelPendingGroupEmailsAction.NONE,
      });

@@ -225,6 +226,7 @@ export class MessageChannelSyncStatusService {
        syncStatus: MessageChannelSyncStatus.ACTIVE,
        syncStage: MessageChannelSyncStage.MESSAGE_LIST_FETCH_PENDING,
        throttleFailureCount: 0,
+        throttleRetryAfter: null,
        syncStageStartedAt: null,
        syncedAt: new Date().toISOString(),
      });
@@ -307,6 +309,7 @@ export class MessageChannelSyncStatusService {
      await messageChannelRepository.update(messageChannelIds, {
        syncStage: MessageChannelSyncStage.FAILED,
        syncStatus: syncStatus,
+        throttleRetryAfter: null,
      });

      const metricsKey =
@@ -98,6 +98,7 @@ export class MessageChannelWorkspaceEntity extends BaseWorkspaceEntity {
  syncStage: MessageChannelSyncStage;
  syncStageStartedAt: string | null;
  throttleFailureCount: number;
+  throttleRetryAfter: string | null;
  connectedAccount: EntityRelation<ConnectedAccountWorkspaceEntity>;
  connectedAccountId: string;
  messageChannelMessageAssociations: EntityRelation<
@@ -3,6 +3,7 @@ import { type MessageNetworkExceptionCode } from 'src/modules/messaging/message-
 export class MessageImportDriverException extends Error {
  code: MessageImportDriverExceptionCode | MessageNetworkExceptionCode;
  cause?: Error;
+  throttleRetryAfter?: Date;
  context?: {
    messageChannelId?: string;
    workspaceId?: string;
@@ -14,6 +15,7 @@ export class MessageImportDriverException extends Error {
    code: MessageImportDriverExceptionCode | MessageNetworkExceptionCode,
    options?: {
      cause?: Error;
+      throttleRetryAfter?: Date;
      context?: {
        messageChannelId?: string;
        workspaceId?: string;
@@ -25,6 +27,7 @@ export class MessageImportDriverException extends Error {
    this.name = 'MessageImportDriverException';
    this.code = code;
    this.cause = options?.cause;
+    this.throttleRetryAfter = options?.throttleRetryAfter;
    this.context = options?.context;

    if (options?.cause?.stack) {
@@ -50,9 +50,11 @@ const ERROR_DEFINITIONS: Record<number, Record<string, ErrorConfig>> = {
 export const getGmailApiError = ({
  code,
  reason,
+  message,
 }: {
  code: number;
  reason?: string;
+  message?: string;
 }): GaxiosError => {
  const statusMap = ERROR_DEFINITIONS[code];

@@ -62,8 +64,10 @@ export const getGmailApiError = ({

  const config = statusMap[reason || ''] ?? statusMap.default;

+  const errorMessage = message ?? config.message;
+
  return new GaxiosError(
-    config.message,
+    errorMessage,
    { url: 'https://gmail.googleapis.com/mocks' },
    {
      status: code,
@@ -71,10 +75,10 @@ export const getGmailApiError = ({
      data: {
        error: {
          code,
-          message: config.message,
+          message: errorMessage,
          errors: [
            {
-              message: config.message,
+              message: errorMessage,
              reason: config.reason,
            },
          ],
@@ -0,0 +1,40 @@
+import { parseGmailErrorRetryAfter } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gmail-error-retry-after.util';
+
+describe('parseGmailErrorRetryAfter', () => {
+  it('should extract the retry-after date from a Gmail 429 error message', () => {
+    const fifteenMinutesFromNow = new Date(Date.now() + 15 * 60 * 1000);
+    const message = `User-rate limit exceeded.  Retry after ${fifteenMinutesFromNow.toISOString()}`;
+
+    const result = parseGmailErrorRetryAfter(message);
+
+    expect(result).toBeInstanceOf(Date);
+    expect(result!.getTime()).toBeCloseTo(fifteenMinutesFromNow.getTime(), -3);
+  });
+
+  it('should return undefined when the message contains no retry-after timestamp', () => {
+    expect(
+      parseGmailErrorRetryAfter('Too Many Concurrent Requests'),
+    ).toBeUndefined();
+  });
+
+  it('should return undefined when the retry-after timestamp has already passed', () => {
+    const fiveMinutesAgo = new Date(Date.now() - 5 * 60 * 1000);
+    const message = `User-rate limit exceeded.  Retry after ${fiveMinutesAgo.toISOString()}`;
+
+    expect(parseGmailErrorRetryAfter(message)).toBeUndefined();
+  });
+
+  it('should return undefined for an empty string', () => {
+    expect(parseGmailErrorRetryAfter('')).toBeUndefined();
+  });
+
+  it('should match case variations of "Retry after"', () => {
+    const fifteenMinutesFromNow = new Date(Date.now() + 15 * 60 * 1000);
+    const message = `User-rate limit exceeded.  retry after ${fifteenMinutesFromNow.toISOString()}`;
+
+    const result = parseGmailErrorRetryAfter(message);
+
+    expect(result).toBeInstanceOf(Date);
+    expect(result!.getTime()).toBeCloseTo(fifteenMinutesFromNow.getTime(), -3);
+  });
+});
@@ -232,4 +232,48 @@ describe('parseGmailApiError', () => {
      MessageImportDriverExceptionCode.SYNC_CURSOR_ERROR,
    );
  });
+
+  it('should populate retryAfter on the exception when a 429 contains a retry-after timestamp', () => {
+    const fifteenMinutesFromNow = new Date(Date.now() + 15 * 60 * 1000);
+    const error = getGmailApiError({
+      code: 429,
+      message: `User-rate limit exceeded.  Retry after ${fifteenMinutesFromNow.toISOString()}`,
+    });
+
+    const exception = parseGmailApiError(error);
+
+    expect(exception.throttleRetryAfter).toBeInstanceOf(Date);
+    expect(exception.throttleRetryAfter!.getTime()).toBeCloseTo(
+      fifteenMinutesFromNow.getTime(),
+      -3,
+    );
+  });
+
+  it('should leave retryAfter undefined on 429 without a retry-after timestamp', () => {
+    const error = getGmailApiError({ code: 429 });
+
+    const exception = parseGmailApiError(error);
+
+    expect(exception.throttleRetryAfter).toBeUndefined();
+  });
+
+  it('should populate retryAfter on the exception when a 403 rateLimitExceeded contains a retry-after timestamp', () => {
+    const fifteenMinutesFromNow = new Date(Date.now() + 15 * 60 * 1000);
+    const error = getGmailApiError({
+      code: 403,
+      reason: 'rateLimit',
+      message: `Rate Limit Exceeded.  Retry after ${fifteenMinutesFromNow.toISOString()}`,
+    });
+
+    const exception = parseGmailApiError(error);
+
+    expect(exception.code).toBe(
+      MessageImportDriverExceptionCode.TEMPORARY_ERROR,
+    );
+    expect(exception.throttleRetryAfter).toBeInstanceOf(Date);
+    expect(exception.throttleRetryAfter!.getTime()).toBeCloseTo(
+      fifteenMinutesFromNow.getTime(),
+      -3,
+    );
+  });
 });
@@ -4,6 +4,7 @@ import {
  MessageImportDriverException,
  MessageImportDriverExceptionCode,
 } from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception';
+import { parseGmailErrorRetryAfter } from 'src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-gmail-error-retry-after.util';

 export const parseGmailApiError = (
  error: GaxiosError,
@@ -57,6 +58,9 @@ export const parseGmailApiError = (
      return new MessageImportDriverException(
        gmailApiError.message,
        MessageImportDriverExceptionCode.TEMPORARY_ERROR,
+        {
+          throttleRetryAfter: parseGmailErrorRetryAfter(gmailApiError.message),
+        },
      );

    case 403:
@@ -68,6 +72,11 @@ export const parseGmailApiError = (
        return new MessageImportDriverException(
          gmailApiError.message,
          MessageImportDriverExceptionCode.TEMPORARY_ERROR,
+          {
+            throttleRetryAfter: parseGmailErrorRetryAfter(
+              gmailApiError.message,
+            ),
+          },
        );
      }
      if (gmailApiError.reason === 'domainPolicy') {
@@ -0,0 +1,24 @@
+const RETRY_AFTER_REGEX =
+  /Retry after (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)/i;
+
+export const parseGmailErrorRetryAfter = (
+  message: string,
+): Date | undefined => {
+  const match = message.match(RETRY_AFTER_REGEX);
+
+  if (!match) {
+    return undefined;
+  }
+
+  const retryAfter = new Date(match[1]);
+
+  if (isNaN(retryAfter.getTime())) {
+    return undefined;
+  }
+
+  if (retryAfter <= new Date()) {
+    return undefined;
+  }
+
+  return retryAfter;
+};
@@ -84,6 +84,7 @@ export class MessagingMessageListFetchJob {
          isThrottled(
            messageChannel.syncStageStartedAt,
            messageChannel.throttleFailureCount,
+            messageChannel.throttleRetryAfter,
          )
        ) {
          await this.messageChannelSyncStatusService.markAsMessagesListFetchPending(
@@ -82,6 +82,7 @@ export class MessagingMessagesImportJob {
        isThrottled(
          messageChannel.syncStageStartedAt,
          messageChannel.throttleFailureCount,
+          messageChannel.throttleRetryAfter,
        )
      ) {
        await this.messageChannelSyncStatusService.markAsMessagesImportPending(
@@ -38,6 +38,7 @@ export class MessagingCursorService {
          },
          {
            throttleFailureCount: 0,
+            throttleRetryAfter: null,
            syncStageStartedAt: null,
            syncCursor:
              !messageChannel.syncCursor ||
@@ -61,6 +62,7 @@ export class MessagingCursorService {
          },
          {
            throttleFailureCount: 0,
+            throttleRetryAfter: null,
            syncStageStartedAt: null,
          },
        );
@@ -1,5 +1,7 @@
 import { Injectable } from '@nestjs/common';

+import { isDefined } from 'twenty-shared/utils';
+
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import {
  type TwentyORMException,
@@ -166,6 +168,20 @@ export class MessageImportExceptionHandlerService {
        undefined,
        ['throttleFailureCount', 'id'],
      );
+
+      const throttleRetryAfter =
+        exception instanceof MessageImportDriverException
+          ? exception.throttleRetryAfter
+          : undefined;
+
+      await messageChannelRepository.update(
+        { id: messageChannel.id },
+        {
+          throttleRetryAfter: isDefined(throttleRetryAfter)
+            ? throttleRetryAfter.toISOString()
+            : null,
+        },
+      );
    }, authContext);

    switch (syncStep) {
@@ -189,6 +189,7 @@ export class MessagingMessagesImportService {
          },
          {
            throttleFailureCount: 0,
+            throttleRetryAfter: null,
            syncStageStartedAt: null,
          },
        );