Add application job enqueue limits guard on all message queue drivers (#23570)

## Context Applications trigger logic functions from several paths (cron, database events, HTTP routes, install/connect hooks). Without a cap, a single app can flood the `logic-function-queue` and starve it. This adds enqueue limits scoped to that queue, mirroring the existing per-application API rate limiting. ## What changed `JobEnqueueThrottlerGuard` reuses the `ThrottlerService` token bucket (same primitive as the API rate limiter) with two tiers: - **Per application installation** (`enqueue:throttler:application:{applicationId}`) - lower ceiling, `APPLICATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT` (default 500). - **Per application registration** (`enqueue:throttler:application-registration:{applicationRegistrationId}`) - higher ceiling shared across all workspaces that installed the same app, `APPLICATION_REGISTRATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT` (default 2000). Both share `APPLICATION_JOB_ENQUEUE_RATE_LIMITING_TTL_IN_MS` (default 60s). Both buckets are checked before either is debited, so a rejection on one tier never burns quota on the other. **Per-queue guarding.** A `ThrottledMessageQueueDriver` decorator wraps the concrete driver (BullMQ or Sync) in the `QUEUE_DRIVER` provider and passes the `queueName` to the guard. The guard only acts on queues in `GUARDED_ENQUEUE_QUEUES` (currently just `logic-function-queue`); every other queue is untouched. **Required application context.** The guard reads a dedicated `applicationJobEnqueueContextStorage` (AsyncLocalStorage) carrying `{ applicationId, applicationRegistrationId }`, and throws if a guarded-queue enqueue runs without both. Every logic-function-queue enqueue site wraps its `add`/`bulkAdd` in `withApplicationJobEnqueueContext`: - cron trigger - database-event trigger (groups logic functions by application, one batch per application) - server route trigger - application post-install hook - connection-provider on-connect hook When the limit is reached the guard records a `JobEnqueueApplicationRateLimited` metric and throws `ThrottlerException` (mapped to 429 by the existing handlers). ## Files - `message-queue/guards/job-enqueue-throttler.guard.ts` - the guard (new) - `message-queue/storage/application-job-enqueue-context.storage.ts` - dedicated enqueue context (new) - `message-queue/constants/guarded-enqueue-queues.constant.ts` - guarded-queue set (new) - `message-queue/drivers/throttled-message-queue.driver.ts` - decorator driver wrapping any driver (new) - `message-queue/message-queue-core.module.ts` - wires the guard into the driver provider - `twenty-config/config-variables.ts` - three tunable `RATE_LIMITING` config variables - `metrics/types/metrics-keys.type.ts` - `JobEnqueueApplicationRateLimited` key - the 5 enqueue sites above - inject the enqueue context ## Notes / trade-offs - A logic function whose application has no `applicationRegistrationId` is skipped at the trigger paths (the install hook throws), matching how the server route trigger already treats "not linked to a registration". - `addCron` is left ungated (idempotent upsert). - Default limits are placeholders and tunable per instance. ## Testing - `JobEnqueueThrottlerGuard` unit tests (7 cases): non-guarded queue skip, throw on missing/partial context, two-tier throttling with distinct limits, per-item token consumption on bulk, no partial debit when either tier is exhausted. - Updated `connection-provider-oauth-flow.service.spec.ts` for the new cache key. - `npx nx typecheck twenty-server` passes; oxlint + oxfmt clean on changed files.
2026-07-30 18:40:22 +02:00
parent cc8fac46c3
commit a99e62fbca
8 changed files with 329 additions and 13 deletions
@@ -572,6 +572,7 @@ export class ApplicationInstallService {
        },
        { retryLimit: 3 },
      );
+
      return;
    }

@@ -5,6 +5,7 @@ import { isDefined } from 'twenty-shared/utils';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
 import { Repository } from 'typeorm';

+import { findActiveFlatApplicationById } from 'src/engine/core-modules/application/utils/find-active-flat-application-by-id.util';
 import { CronTriggerDeduplicationService } from 'src/engine/core-modules/cron/services/cron-trigger-deduplication.service';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
@@ -12,6 +13,7 @@ import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decora
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { ApplicationJobEnqueueThrottlerService } from 'src/engine/core-modules/message-queue/services/application-job-enqueue-throttler.service';
 import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import {
@@ -34,6 +36,7 @@ export class CronTriggerCronJob {
    private readonly workspaceCacheService: WorkspaceCacheService,
    private readonly exceptionHandlerService: ExceptionHandlerService,
    private readonly cronTriggerDeduplicationService: CronTriggerDeduplicationService,
+    private readonly applicationJobEnqueueThrottlerService: ApplicationJobEnqueueThrottlerService,
  ) {}

  @Process(CronTriggerCronJob.name)
@@ -50,9 +53,10 @@ export class CronTriggerCronJob {

    for (const activeWorkspace of activeWorkspaces) {
      try {
-        const { flatLogicFunctionMaps } =
+        const { flatLogicFunctionMaps, flatApplicationMaps } =
          await this.workspaceCacheService.getOrRecompute(activeWorkspace.id, [
            'flatLogicFunctionMaps',
+            'flatApplicationMaps',
          ]);

        const logicFunctions = Object.values(
@@ -85,6 +89,22 @@ export class CronTriggerCronJob {
            continue;
          }

+          const application = findActiveFlatApplicationById(
+            flatApplicationMaps,
+            logicFunction.applicationId,
+          );
+          const applicationRegistrationId =
+            application?.applicationRegistrationId;
+
+          if (!isDefined(applicationRegistrationId)) {
+            continue;
+          }
+
+          await this.applicationJobEnqueueThrottlerService.throttleOrThrow({
+            applicationId: logicFunction.applicationId,
+            applicationRegistrationId,
+          });
+
          await this.messageQueueService.add<LogicFunctionTriggerJobData>(
            LogicFunctionTriggerJob.name,
            {
@@ -2,10 +2,12 @@ import { isDefined } from 'twenty-shared/utils';

 import type { ObjectRecordEvent } from 'twenty-shared/database-events';

+import { findActiveFlatApplicationById } from 'src/engine/core-modules/application/utils/find-active-flat-application-by-id.util';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { ApplicationJobEnqueueThrottlerService } from 'src/engine/core-modules/message-queue/services/application-job-enqueue-throttler.service';
 import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { transformEventBatchToEventPayloads } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/database-event/utils/transform-event-batch-to-event-payloads';
 import {
@@ -21,14 +23,15 @@ export class CallDatabaseEventTriggerJobsJob {
    @InjectMessageQueue(MessageQueue.logicFunctionQueue)
    private readonly messageQueueService: MessageQueueService,
    private readonly workspaceCacheService: WorkspaceCacheService,
+    private readonly applicationJobEnqueueThrottlerService: ApplicationJobEnqueueThrottlerService,
  ) {}

  @Process(CallDatabaseEventTriggerJobsJob.name)
  async handle(workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>) {
-    const { flatLogicFunctionMaps } =
+    const { flatLogicFunctionMaps, flatApplicationMaps } =
      await this.workspaceCacheService.getOrRecompute(
        workspaceEventBatch.workspaceId,
-        ['flatLogicFunctionMaps'],
+        ['flatLogicFunctionMaps', 'flatApplicationMaps'],
      );

    const logicFunctionsWithDatabaseEventTrigger = Object.values(
@@ -51,16 +54,57 @@ export class CallDatabaseEventTriggerJobsJob {
        }),
      );

-    const logicFunctionPayloads = transformEventBatchToEventPayloads({
-      logicFunctions: logicFunctionsToTrigger,
-      workspaceEventBatch,
-    });
+    const logicFunctionsByApplicationId = new Map<
+      string,
+      typeof logicFunctionsToTrigger
+    >();

-    await this.messageQueueService.bulkAdd<LogicFunctionTriggerJobData>(
-      LogicFunctionTriggerJob.name,
-      logicFunctionPayloads,
-      { retryLimit: 3 },
-    );
+    for (const logicFunction of logicFunctionsToTrigger) {
+      const applicationLogicFunctions =
+        logicFunctionsByApplicationId.get(logicFunction.applicationId) ?? [];
+
+      applicationLogicFunctions.push(logicFunction);
+      logicFunctionsByApplicationId.set(
+        logicFunction.applicationId,
+        applicationLogicFunctions,
+      );
+    }
+
+    for (const [
+      applicationId,
+      logicFunctions,
+    ] of logicFunctionsByApplicationId) {
+      const application = findActiveFlatApplicationById(
+        flatApplicationMaps,
+        applicationId,
+      );
+      const applicationRegistrationId = application?.applicationRegistrationId;
+
+      if (!isDefined(applicationRegistrationId)) {
+        continue;
+      }
+
+      const logicFunctionPayloads = transformEventBatchToEventPayloads({
+        logicFunctions,
+        workspaceEventBatch,
+      });
+
+      if (logicFunctionPayloads.length === 0) {
+        continue;
+      }
+
+      await this.applicationJobEnqueueThrottlerService.throttleOrThrow({
+        applicationId,
+        applicationRegistrationId,
+        jobCount: logicFunctionPayloads.length,
+      });
+
+      await this.messageQueueService.bulkAdd<LogicFunctionTriggerJobData>(
+        LogicFunctionTriggerJob.name,
+        logicFunctionPayloads,
+        { retryLimit: 3 },
+      );
+    }
  }

  private shouldTriggerJob({
@@ -20,9 +20,11 @@ import {
  ConfigurableModuleClass,
  type OPTIONS_TYPE,
 } from 'src/engine/core-modules/message-queue/message-queue.module-definition';
+import { ApplicationJobEnqueueThrottlerService } from 'src/engine/core-modules/message-queue/services/application-job-enqueue-throttler.service';
 import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { getQueueToken } from 'src/engine/core-modules/message-queue/utils/get-queue-token.util';
 import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
+import { ThrottlerModule } from 'src/engine/core-modules/throttler/throttler.module';

@Global()
@Module({})
@@ -43,13 +45,20 @@ export class MessageQueueCoreModule extends ConfigurableModuleClass {

    return {
      ...dynamicModule,
+      imports: [
+        ...(dynamicModule.imports ?? []),
+        MetricsModule,
+        ThrottlerModule,
+      ],
      providers: [
        ...(dynamicModule.providers ?? []),
+        ApplicationJobEnqueueThrottlerService,
        driverProvider,
        ...queueProviders,
      ],
      exports: [
        ...(dynamicModule.exports ?? []),
+        ApplicationJobEnqueueThrottlerService,
        ...Object.values(MessageQueue).map((queueName) =>
          getQueueToken(queueName),
        ),
@@ -78,14 +87,20 @@ export class MessageQueueCoreModule extends ConfigurableModuleClass {

    return {
      ...dynamicModule,
-      imports: [...(dynamicModule.imports ?? []), MetricsModule],
+      imports: [
+        ...(dynamicModule.imports ?? []),
+        MetricsModule,
+        ThrottlerModule,
+      ],
      providers: [
        ...(dynamicModule.providers ?? []),
+        ApplicationJobEnqueueThrottlerService,
        driverProvider,
        ...queueProviders,
      ],
      exports: [
        ...(dynamicModule.exports ?? []),
+        ApplicationJobEnqueueThrottlerService,
        ...Object.values(MessageQueue).map((queueName) =>
          getQueueToken(queueName),
        ),
@@ -0,0 +1,123 @@
+import { Test, type TestingModule } from '@nestjs/testing';
+
+import { ApplicationJobEnqueueThrottlerService } from 'src/engine/core-modules/message-queue/services/application-job-enqueue-throttler.service';
+import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
+import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
+import { ThrottlerException } from 'src/engine/core-modules/throttler/throttler.exception';
+import { ThrottlerService } from 'src/engine/core-modules/throttler/throttler.service';
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+
+const CONFIG = {
+  APPLICATION_JOB_ENQUEUE_RATE_LIMITING_TTL_IN_MS: 60_000,
+  APPLICATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT: 500,
+  APPLICATION_REGISTRATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT: 2000,
+} as const;
+
+const context = {
+  applicationId: 'application-id',
+  applicationRegistrationId: 'registration-id',
+};
+
+const APPLICATION_KEY = 'enqueue:throttler:application:application-id';
+const REGISTRATION_KEY =
+  'enqueue:throttler:application-registration:registration-id';
+
+describe('ApplicationJobEnqueueThrottlerService', () => {
+  let service: ApplicationJobEnqueueThrottlerService;
+  let throttlerService: {
+    getAvailableTokensCount: jest.Mock;
+    consumeTokens: jest.Mock;
+  };
+  let metricsService: { incrementCounterForEvent: jest.Mock };
+
+  const mockAvailableTokens = (byKey: Record<string, number>) => {
+    throttlerService.getAvailableTokensCount.mockImplementation(
+      async (key: string) => byKey[key] ?? 0,
+    );
+  };
+
+  beforeEach(async () => {
+    throttlerService = {
+      getAvailableTokensCount: jest.fn().mockResolvedValue(1000),
+      consumeTokens: jest.fn().mockResolvedValue(undefined),
+    };
+    metricsService = {
+      incrementCounterForEvent: jest.fn().mockResolvedValue(undefined),
+    };
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        ApplicationJobEnqueueThrottlerService,
+        { provide: ThrottlerService, useValue: throttlerService },
+        { provide: MetricsService, useValue: metricsService },
+        {
+          provide: TwentyConfigService,
+          useValue: {
+            get: (key: keyof typeof CONFIG) => CONFIG[key],
+          },
+        },
+      ],
+    }).compile();
+
+    service = module.get(ApplicationJobEnqueueThrottlerService);
+  });
+
+  it('debits both application and registration buckets with distinct limits', async () => {
+    await service.throttleOrThrow(context);
+
+    expect(throttlerService.consumeTokens).toHaveBeenCalledWith(
+      APPLICATION_KEY,
+      1,
+      CONFIG.APPLICATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT,
+      CONFIG.APPLICATION_JOB_ENQUEUE_RATE_LIMITING_TTL_IN_MS,
+    );
+    expect(throttlerService.consumeTokens).toHaveBeenCalledWith(
+      REGISTRATION_KEY,
+      1,
+      CONFIG.APPLICATION_REGISTRATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT,
+      CONFIG.APPLICATION_JOB_ENQUEUE_RATE_LIMITING_TTL_IN_MS,
+    );
+  });
+
+  it('consumes one token per job for bulk enqueue', async () => {
+    await service.throttleOrThrow({ ...context, jobCount: 5 });
+
+    expect(throttlerService.consumeTokens).toHaveBeenCalledWith(
+      APPLICATION_KEY,
+      5,
+      expect.any(Number),
+      expect.any(Number),
+    );
+  });
+
+  it('does not debit any bucket when the registration bucket is exhausted', async () => {
+    mockAvailableTokens({
+      [APPLICATION_KEY]: 1000,
+      [REGISTRATION_KEY]: 0,
+    });
+
+    await expect(service.throttleOrThrow(context)).rejects.toThrow(
+      ThrottlerException,
+    );
+
+    expect(throttlerService.consumeTokens).not.toHaveBeenCalled();
+    expect(metricsService.incrementCounterForEvent).toHaveBeenCalledWith(
+      expect.objectContaining({
+        key: MetricsKeys.JobEnqueueApplicationRateLimited,
+      }),
+    );
+  });
+
+  it('does not debit any bucket when the application bucket is exhausted', async () => {
+    mockAvailableTokens({
+      [APPLICATION_KEY]: 0,
+      [REGISTRATION_KEY]: 1000,
+    });
+
+    await expect(service.throttleOrThrow(context)).rejects.toThrow(
+      ThrottlerException,
+    );
+
+    expect(throttlerService.consumeTokens).not.toHaveBeenCalled();
+  });
+});
@@ -0,0 +1,85 @@
+import { Injectable } from '@nestjs/common';
+
+import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service';
+import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type';
+import {
+  ThrottlerException,
+  ThrottlerExceptionCode,
+} from 'src/engine/core-modules/throttler/throttler.exception';
+import { ThrottlerService } from 'src/engine/core-modules/throttler/throttler.service';
+import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service';
+
+@Injectable()
+export class ApplicationJobEnqueueThrottlerService {
+  constructor(
+    private readonly throttlerService: ThrottlerService,
+    private readonly twentyConfigService: TwentyConfigService,
+    private readonly metricsService: MetricsService,
+  ) {}
+
+  async throttleOrThrow({
+    applicationId,
+    applicationRegistrationId,
+    jobCount = 1,
+  }: {
+    applicationId: string;
+    applicationRegistrationId: string;
+    jobCount?: number;
+  }): Promise<void> {
+    const timeWindow = this.twentyConfigService.get(
+      'APPLICATION_JOB_ENQUEUE_RATE_LIMITING_TTL_IN_MS',
+    );
+    const applicationKey = `enqueue:throttler:application:${applicationId}`;
+    const applicationLimit = this.twentyConfigService.get(
+      'APPLICATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT',
+    );
+    const registrationKey = `enqueue:throttler:application-registration:${applicationRegistrationId}`;
+    const registrationLimit = this.twentyConfigService.get(
+      'APPLICATION_REGISTRATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT',
+    );
+
+    const [applicationTokens, registrationTokens] = await Promise.all([
+      this.throttlerService.getAvailableTokensCount(
+        applicationKey,
+        applicationLimit,
+        timeWindow,
+      ),
+      this.throttlerService.getAvailableTokensCount(
+        registrationKey,
+        registrationLimit,
+        timeWindow,
+      ),
+    ]);
+
+    if (applicationTokens < jobCount || registrationTokens < jobCount) {
+      await this.metricsService.incrementCounterForEvent({
+        key: MetricsKeys.JobEnqueueApplicationRateLimited,
+        shouldStoreInCache: false,
+        attributes: {
+          application_id: applicationId,
+          application_registration_id: applicationRegistrationId,
+        },
+      });
+
+      throw new ThrottlerException(
+        'Application job enqueue limit reached',
+        ThrottlerExceptionCode.LIMIT_REACHED,
+      );
+    }
+
+    await Promise.all([
+      this.throttlerService.consumeTokens(
+        applicationKey,
+        jobCount,
+        applicationLimit,
+        timeWindow,
+      ),
+      this.throttlerService.consumeTokens(
+        registrationKey,
+        jobCount,
+        registrationLimit,
+        timeWindow,
+      ),
+    ]);
+  }
+}
@@ -59,6 +59,7 @@ export enum MetricsKeys {
  SignUpSuccess = 'sign-up/success',
  CommonApiQueryRateLimited = 'common-api-query/rate-limited',
  CommonApiApplicationQueryRateLimited = 'common-api-query/application-rate-limited',
+  JobEnqueueApplicationRateLimited = 'job/enqueue-application-rate-limited',
  JobCompleted = 'job/completed',
  JobFailed = 'job/failed',
  JobStalled = 'job/stalled',
@@ -1521,6 +1521,33 @@ export class ConfigVariables {
  @CastToPositiveNumber()
  APPLICATION_API_RATE_LIMITING_LIMIT = 500;

+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.RATE_LIMITING,
+    description:
+      'Time-to-live for application job enqueue rate limiting in milliseconds',
+    type: ConfigVariableType.NUMBER,
+  })
+  @CastToPositiveNumber()
+  APPLICATION_JOB_ENQUEUE_RATE_LIMITING_TTL_IN_MS = 60_000;
+
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.RATE_LIMITING,
+    description:
+      'Maximum number of jobs a single application installation can enqueue in the rate limiting window',
+    type: ConfigVariableType.NUMBER,
+  })
+  @CastToPositiveNumber()
+  APPLICATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT = 500;
+
+  @ConfigVariablesMetadata({
+    group: ConfigVariablesGroup.RATE_LIMITING,
+    description:
+      'Maximum number of jobs enqueued per application registration across all workspaces in the rate limiting window',
+    type: ConfigVariableType.NUMBER,
+  })
+  @CastToPositiveNumber()
+  APPLICATION_REGISTRATION_JOB_ENQUEUE_RATE_LIMITING_LIMIT = 2000;
+
  @CastToPositiveNumber()
  @ConfigVariablesMetadata({
    group: ConfigVariablesGroup.RATE_LIMITING,