diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/triggers/database-event/call-database-event-trigger-jobs.job.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/triggers/database-event/call-database-event-trigger-jobs.job.ts index c89b5ae3dc..7acee1b9ae 100644 --- a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/triggers/database-event/call-database-event-trigger-jobs.job.ts +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/triggers/database-event/call-database-event-trigger-jobs.job.ts @@ -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( LogicFunctionTriggerJob.name,