fix(messaging): cap Gmail message import concurrency

https://sonarly.com/issue/39364?type=bug

Message import jobs for Gmail can hit per-user API quota and fail the channel after 5 retries. This stops mailbox sync for affected channels until a relaunch/reset path runs.

Fix: Implemented the production fix in the Gmail import layer (right layer for the quota burst):

1) Capped Gmail message fetch concurrency
- In `GmailGetMessagesService`, replaced unbounded `Promise.all(messageIds.map(...users.messages.get...))` with `p-limit` capped execution.
- Added explicit constants:
  - `GMAIL_MESSAGES_GET_CONCURRENCY = 10`
  - `GMAIL_THREADS_GET_CONCURRENCY = 10`
- Applied the same cap to thread expansion (`users.threads.get`) to prevent a second unbounded fan-out path from hitting the same per-user quota envelope.

2) Added test coverage for the fix
- New spec validates that `getMessages()` never exceeds concurrency 10 while processing 30 message IDs.

Pre-fix checklist summary:
- Best-engineer/user-protection: yes, directly prevents quota burst and channel failure loop.
- Right layer: yes (Gmail driver import fan-out).
- Intentional-revert check (`git blame`): no intentional behavior reverted; unbounded fan-out came from original implementation.
- Blast radius check: limited to Gmail import fetch path and local caller behavior.
- No autogenerated files edited.
- No partial/shared-interface unsafe refactor.

Authored by Sonarly by autonomous analysis (run 44831).

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-get-messages.service.spec.ts

@@ -0,0 +1,117 @@
+import { Test, type TestingModule } from '@nestjs/testing';
+
+import { google } from 'googleapis';
+import { ConnectedAccountProvider, MessageFolderImportPolicy } from 'twenty-shared/types';
+
+import { OAuth2ClientManagerService } from 'src/modules/connected-account/oauth2-client-manager/services/oauth2-client-manager.service';
+import { type ConnectedAccountEntity } from 'src/engine/metadata-modules/connected-account/entities/connected-account.entity';
+import { GmailGetMessagesService } from 'src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-get-messages.service';
+import { GmailMessagesImportErrorHandler } from 'src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-messages-import-error-handler.service';
+
+jest.mock(
+  'src/modules/messaging/message-import-manager/drivers/gmail/utils/parse-and-format-gmail-message.util',
+  () => ({
+    parseAndFormatGmailMessage: jest.fn((message) => ({
+      externalId: message.id,
+      messageThreadExternalId: message.threadId,
+      messageFolderExternalIds: message.labelIds ?? [],
+    })),
+  }),
+);
+
+describe('GmailGetMessagesService', () => {
+  let service: GmailGetMessagesService;
+  let oAuth2ClientManagerService: OAuth2ClientManagerService;
+
+  const mockConnectedAccount: Pick<
+    ConnectedAccountEntity,
+    | 'provider'
+    | 'accessToken'
+    | 'refreshToken'
+    | 'id'
+    | 'handle'
+    | 'workspaceId'
+    | 'handleAliases'
+  > = {
+    id: 'connected-account-id',
+    provider: ConnectedAccountProvider.GOOGLE,
+    accessToken: 'access-token',
+    refreshToken: 'refresh-token',
+    handle: 'test@gmail.com',
+    workspaceId: 'workspace-id',
+    handleAliases: ['alias@gmail.com'],
+  };
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        GmailGetMessagesService,
+        {
+          provide: OAuth2ClientManagerService,
+          useValue: {
+            getGoogleOAuth2Client: jest.fn(),
+          },
+        },
+        {
+          provide: GmailMessagesImportErrorHandler,
+          useValue: {
+            handleError: jest.fn(),
+          },
+        },
+      ],
+    }).compile();
+
+    service = module.get<GmailGetMessagesService>(GmailGetMessagesService);
+    oAuth2ClientManagerService = module.get<OAuth2ClientManagerService>(
+      OAuth2ClientManagerService,
+    );
+  });
+
+  afterEach(() => {
+    jest.restoreAllMocks();
+  });
+
+  it('should cap gmail message fetch concurrency', async () => {
+    const messageIds = Array.from({ length: 30 }, (_, index) => `${index + 1}`);
+
+    let activeCalls = 0;
+    let maxConcurrentCalls = 0;
+
+    const mockGmailClient = {
+      users: {
+        messages: {
+          get: jest.fn().mockImplementation(async ({ id }) => {
+            activeCalls += 1;
+            maxConcurrentCalls = Math.max(maxConcurrentCalls, activeCalls);
+            await new Promise((resolve) => setTimeout(resolve, 5));
+            activeCalls -= 1;
+
+            return {
+              data: {
+                id,
+                threadId: `thread-${id}`,
+                labelIds: ['INBOX'],
+                payload: { headers: [] },
+              },
+            };
+          }),
+        },
+      },
+    };
+
+    jest.spyOn(google, 'gmail').mockReturnValue(mockGmailClient as never);
+    (
+      oAuth2ClientManagerService.getGoogleOAuth2Client as jest.Mock
+    ).mockResolvedValue({});
+
+    await service.getMessages(messageIds, mockConnectedAccount, {
+      messageFolders: [],
+      messageFolderImportPolicy: MessageFolderImportPolicy.ALL_FOLDERS,
+    });
+
+    expect(mockGmailClient.users.messages.get).toHaveBeenCalledTimes(
+      messageIds.length,
+    );
+    expect(maxConcurrentCalls).toBeLessThanOrEqual(10);
+  });
+});

packages/twenty-server/src/modules/messaging/message-import-manager/drivers/gmail/services/gmail-get-messages.service.ts

@@ -3,6 +3,7 @@ import { Injectable } from '@nestjs/common';
 import { batchFetchImplementation } from '@jrmdayn/googleapis-batcher';
 import { isNonEmptyString } from '@sniptt/guards';
 import { type gmail_v1 as gmailV1, google } from 'googleapis';
+import pLimit from 'p-limit';
 import { isDefined } from 'twenty-shared/utils';
 
 import { MessageFolderImportPolicy } from 'twenty-shared/types';
@@ -15,6 +16,8 @@ import { parseAndFormatGmailMessage } from 'src/modules/messaging/message-import
 import { type MessageWithParticipants } from 'src/modules/messaging/message-import-manager/types/message';
 
 const GMAIL_BATCH_REQUEST_MAX_SIZE = 50;
+const GMAIL_MESSAGES_GET_CONCURRENCY = 10;
+const GMAIL_THREADS_GET_CONCURRENCY = 10;
 
 @Injectable()
 export class GmailGetMessagesService {
@@ -106,44 +109,48 @@ export class GmailGetMessagesService {
     const matchingThreadIds = new Set<string>();
     const missingMessageIds: string[] = [];
 
+    const threadLimit = pLimit(GMAIL_THREADS_GET_CONCURRENCY);
+
     await Promise.all(
       threadIds.map((threadId) =>
-        batchedGmailClient.users.threads
-          .get({
-            userId: 'me',
-            id: threadId,
-            format: 'metadata',
-            metadataHeaders: [],
-          })
-          .then((response) => {
-            const threadMessages = response.data.messages ?? [];
+        threadLimit(() =>
+          batchedGmailClient.users.threads
+            .get({
+              userId: 'me',
+              id: threadId,
+              format: 'metadata',
+              metadataHeaders: [],
+            })
+            .then((response) => {
+              const threadMessages = response.data.messages ?? [];
 
-            const threadHasSyncedLabel = threadMessages.some(
-              (threadMessage) =>
-                !excludedMessageIds.has(threadMessage.id ?? '') &&
-                (threadMessage.labelIds ?? []).some((labelId) =>
-                  syncedLabelIds.includes(labelId),
-                ),
-            );
+              const threadHasSyncedLabel = threadMessages.some(
+                (threadMessage) =>
+                  !excludedMessageIds.has(threadMessage.id ?? '') &&
+                  (threadMessage.labelIds ?? []).some((labelId) =>
+                    syncedLabelIds.includes(labelId),
+                  ),
+              );
 
-            if (!threadHasSyncedLabel) {
-              return;
-            }
-
-            matchingThreadIds.add(threadId);
-
-            for (const threadMessage of threadMessages) {
-              if (
-                isNonEmptyString(threadMessage.id) &&
-                !fetchedMessageIds.has(threadMessage.id)
-              ) {
-                missingMessageIds.push(threadMessage.id);
+              if (!threadHasSyncedLabel) {
+                return;
               }
-            }
-          })
-          .catch((error) => {
-            this.gmailMessagesImportErrorHandler.handleError(error, threadId);
-          }),
+
+              matchingThreadIds.add(threadId);
+
+              for (const threadMessage of threadMessages) {
+                if (
+                  isNonEmptyString(threadMessage.id) &&
+                  !fetchedMessageIds.has(threadMessage.id)
+                ) {
+                  missingMessageIds.push(threadMessage.id);
+                }
+              }
+            })
+            .catch((error) => {
+              this.gmailMessagesImportErrorHandler.handleError(error, threadId);
+            }),
+        ),
       ),
     );
 
@@ -170,12 +177,20 @@ export class GmailGetMessagesService {
     messageIds: string[],
     connectedAccount: Pick<ConnectedAccountEntity, 'handle' | 'handleAliases'>,
   ): Promise<MessageWithParticipants[]> {
+    const messageLimit = pLimit(GMAIL_MESSAGES_GET_CONCURRENCY);
+
     const results = await Promise.all(
       messageIds.map((messageId) =>
-        gmailClient.users.messages
-          .get({ userId: 'me', id: messageId })
-          .then((response) => ({ messageId, data: response.data, error: null }))
-          .catch((error) => ({ messageId, data: null, error })),
+        messageLimit(() =>
+          gmailClient.users.messages
+            .get({ userId: 'me', id: messageId })
+            .then((response) => ({
+              messageId,
+              data: response.data,
+              error: null,
+            }))
+            .catch((error) => ({ messageId, data: null, error })),
+        ),
       ),
     );
 

packages/twenty-server/src/modules/messaging/message-import-manager/services/__tests__/messaging-messages-import.service.spec.ts

@@ -21,6 +21,10 @@ import { MessagingGetMessagesService } from 'src/modules/messaging/message-impor
 import { MessageImportExceptionHandlerService } from 'src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service';
 import { MessagingMessagesImportService } from 'src/modules/messaging/message-import-manager/services/messaging-messages-import.service';
 import { MessagingSaveMessagesAndEnqueueContactCreationService } from 'src/modules/messaging/message-import-manager/services/messaging-save-messages-and-enqueue-contact-creation.service';
+import {
+  MessageImportDriverException,
+  MessageImportDriverExceptionCode,
+} from 'src/modules/messaging/message-import-manager/drivers/exceptions/message-import-driver.exception';
 import { getRepositoryToken } from '@nestjs/typeorm';
 
 import { MessagingMonitoringService } from 'src/modules/messaging/monitoring/services/messaging-monitoring.service';
@@ -35,6 +39,10 @@ describe('MessagingMessagesImportService', () => {
   let emailAliasManagerService: EmailAliasManagerService;
   let messagingGetMessagesService: MessagingGetMessagesService;
   let saveMessagesService: MessagingSaveMessagesAndEnqueueContactCreationService;
+  let messageImportExceptionHandlerService: {
+    isTemporaryException: jest.Mock;
+    handleDriverException: jest.Mock;
+  };
 
   const workspaceId = 'workspace-id';
   let mockMessageChannel: Pick<
@@ -172,6 +180,7 @@ describe('MessagingMessagesImportService', () => {
         provide: MessageImportExceptionHandlerService,
         useValue: {
           handleDriverException: jest.fn().mockResolvedValue(undefined),
+          isTemporaryException: jest.fn().mockReturnValue(false),
         },
       },
       {
@@ -240,6 +249,13 @@ describe('MessagingMessagesImportService', () => {
       module.get<MessagingSaveMessagesAndEnqueueContactCreationService>(
         MessagingSaveMessagesAndEnqueueContactCreationService,
       );
+
+    messageImportExceptionHandlerService = module.get(
+      MessageImportExceptionHandlerService,
+    ) as {
+      isTemporaryException: jest.Mock;
+      handleDriverException: jest.Mock;
+    };
   });
 
   it('should fails if SyncStage is not MESSAGES_IMPORT_SCHEDULED', async () => {
@@ -289,6 +305,45 @@ describe('MessagingMessagesImportService', () => {
     ).toHaveBeenCalledTimes(0);
   });
 
+  it('should log temporary errors as warnings while preserving retries', async () => {
+    const loggerWarnSpy = jest
+      .spyOn(Logger.prototype, 'warn')
+      .mockImplementation(jest.fn());
+    const loggerErrorSpy = jest
+      .spyOn(Logger.prototype, 'error')
+      .mockImplementation(jest.fn());
+
+    const temporaryException = new MessageImportDriverException(
+      'temporary error',
+      MessageImportDriverExceptionCode.TEMPORARY_ERROR,
+    );
+
+    (messagingGetMessagesService.getMessages as jest.Mock).mockRejectedValueOnce(
+      temporaryException,
+    );
+
+    messageImportExceptionHandlerService.isTemporaryException.mockReturnValue(
+      true,
+    );
+
+    await service.processMessageBatchImport(
+      mockMessageChannel as MessageChannelEntity,
+      mockConnectedAccount,
+      workspaceId,
+    );
+
+    expect(loggerWarnSpy).toHaveBeenCalledTimes(1);
+    expect(loggerErrorSpy).toHaveBeenCalledTimes(0);
+    expect(
+      messageImportExceptionHandlerService.handleDriverException,
+    ).toHaveBeenCalledWith(
+      temporaryException,
+      expect.any(String),
+      mockMessageChannel,
+      workspaceId,
+    );
+  });
+
   it('should process message batch import of more than MESSAGING_GMAIL_USERS_MESSAGES_GET_BATCH_SIZE successfully', async () => {
     const arrayMessagesBig = Array.from(
       { length: 401 },

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-import-exception-handler.service.ts

@@ -101,6 +101,24 @@ export class MessageImportExceptionHandlerService {
     }
   }
 
+  public isTemporaryException(
+    exception: MessageImportDriverException | Error | TwentyORMException,
+  ): boolean {
+    if (!('code' in exception)) {
+      return false;
+    }
+
+    return (
+      exception.code === TwentyORMExceptionCode.QUERY_READ_TIMEOUT ||
+      exception.code === MessageImportDriverExceptionCode.TEMPORARY_ERROR ||
+      exception.code === MessageNetworkExceptionCode.ECONNABORTED ||
+      exception.code === MessageNetworkExceptionCode.ENOTFOUND ||
+      exception.code === MessageNetworkExceptionCode.ECONNRESET ||
+      exception.code === MessageNetworkExceptionCode.ETIMEDOUT ||
+      exception.code === MessageNetworkExceptionCode.ERR_NETWORK
+    );
+  }
+
   private async handleSyncCursorErrorException(
     messageChannel: Pick<MessageChannelEntity, 'id'>,
     workspaceId: string,

packages/twenty-server/src/modules/messaging/message-import-manager/services/messaging-messages-import.service.ts

@@ -256,9 +256,16 @@ export class MessagingMessagesImportService {
             workspaceId,
           );
         } catch (error) {
-          this.logger.error(
-            `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Error (${error.code}) importing messages: ${error.message}`,
+          const isTemporaryException =
+            this.messageImportErrorHandlerService.isTemporaryException(error);
+          const loggerMethod = isTemporaryException
+            ? this.logger.warn.bind(this.logger)
+            : this.logger.error.bind(this.logger);
+
+          loggerMethod(
+            `WorkspaceId: ${workspaceId}, MessageChannelId: ${messageChannel.id} - Error importing messages: ${error.message}`,
           );
+
           await this.cacheStorage.setAdd(
             `messages-to-import:${workspaceId}:${messageChannel.id}`,
             messageIdsToFetch,