fix(server): catch application job enqueue throttle instead of failing the batch job (#23684)
## Context Sentry issue [TWENTY-SERVER-GXJ](https://twenty-v7.sentry.io/issues/7495161692?project=4507072499810304) (`Application job enqueue limit reached`, 10k+ events, 33 workspaces) is the application-job enqueue throttle firing as designed, but being reported as an error. ## Root cause `CallDatabaseEventTriggerJobsJob.handle` calls `throttleOrThrow` inside its per-application loop with no `try/catch`. When an application exceeds its enqueue budget, the `ThrottlerException` propagates out of the bullmq job handler, where `shouldCaptureException` captures it (it has no `statusCode < 500`, so it is not filtered) and the whole batch job fails. Two consequences: - Expected throttling shows up in Sentry as an error (noise). A metric (`JobEnqueueApplicationRateLimited`) already tracks it. - The batch job fails and is retried (`retryLimit: 3`), re-running the loop from the top and re-enqueuing logic-function jobs for applications that already succeeded before the throttled one (duplicate triggers); after retries are exhausted the remaining applications' triggers are dropped. The workflow hard-throttle uses the same `ThrottlerException` but does not show up in Sentry because its call site (`checkHardThrottleLimit`) catches it and turns it into a graceful signal. This PR applies the same pattern to the application enqueue path. ## Change - Wrap the `throttleOrThrow` call: on `ThrottlerException`, `continue` to the next application instead of failing the job; rethrow anything else. - Add a unit test covering the skip-throttled-application and rethrow-other-errors behavior. <!-- This is an auto-generated description by cubic. --> <a href="https://cubic.dev/pr/twentyhq/twenty/pull/23684?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a> <!-- End of auto-generated description by cubic. -->
This commit is contained in:
+22
-5
@@ -1,3 +1,5 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
|
||||
import type { ObjectRecordEvent } from 'twenty-shared/database-events';
|
||||
@@ -9,6 +11,7 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc
|
||||
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 { ThrottlerException } from 'src/engine/core-modules/throttler/throttler.exception';
|
||||
import { transformEventBatchToEventPayloads } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/database-event/utils/transform-event-batch-to-event-payloads';
|
||||
import {
|
||||
LogicFunctionTriggerJob,
|
||||
@@ -19,6 +22,8 @@ import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/wo
|
||||
|
||||
@Processor(MessageQueue.triggerQueue)
|
||||
export class CallDatabaseEventTriggerJobsJob {
|
||||
private readonly logger = new Logger(CallDatabaseEventTriggerJobsJob.name);
|
||||
|
||||
constructor(
|
||||
@InjectMessageQueue(MessageQueue.logicFunctionQueue)
|
||||
private readonly messageQueueService: MessageQueueService,
|
||||
@@ -93,11 +98,23 @@ export class CallDatabaseEventTriggerJobsJob {
|
||||
continue;
|
||||
}
|
||||
|
||||
await this.applicationJobEnqueueThrottlerService.throttleOrThrow({
|
||||
applicationId,
|
||||
applicationRegistrationId,
|
||||
jobCount: logicFunctionPayloads.length,
|
||||
});
|
||||
try {
|
||||
await this.applicationJobEnqueueThrottlerService.throttleOrThrow({
|
||||
applicationId,
|
||||
applicationRegistrationId,
|
||||
jobCount: logicFunctionPayloads.length,
|
||||
});
|
||||
} catch (error) {
|
||||
if (error instanceof ThrottlerException) {
|
||||
this.logger.warn(
|
||||
`Enqueue throttled for application ${applicationId} (registration ${applicationRegistrationId}) in workspace ${workspaceEventBatch.workspaceId}: skipping ${logicFunctionPayloads.length} logic function trigger(s)`,
|
||||
);
|
||||
|
||||
continue;
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
|
||||
await this.messageQueueService.bulkAdd<LogicFunctionTriggerJobData>(
|
||||
LogicFunctionTriggerJob.name,
|
||||
|
||||
Reference in New Issue
Block a user