From a0f196e8713b395fb0ef892cbd407b78d1cea291 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Wed, 3 Dec 2025 10:01:36 +0100 Subject: [PATCH] Improve workflow throttling logic (#16260) - if >5000 workflows per hour, new ones should failed - if >100 workflow per min, new ones should be set as not started. Except manual trigger - when enqueued, we check if there a not started workflows that may be queued. If yes, we call the associated job --------- Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --- .../commands/cron-register-all.command.ts | 6 +- .../metrics/types/metrics-keys.type.ts | 2 +- .../throttler/throttler.service.ts | 62 ++++-- .../twenty-config/config-variables.ts | 28 ++- .../delay/jobs/resume-delayed-workflow.job.ts | 6 - .../workflow-executor.module.ts | 4 +- ...orkflow-executor.workspace-service.spec.ts | 13 -- .../workflow-executor.workspace-service.ts | 5 - .../workflow-runner/jobs/run-workflow.job.ts | 41 +--- .../command/workflow-run-enqueue.command.ts | 39 ---- ...fault-workflow-run-queue-throttle-limit.ts | 1 - .../workflow-run-enqueue.cron.command.ts | 4 +- .../jobs/workflow-run-enqueue.cron.job.ts | 11 +- .../jobs/workflow-run-enqueue.job.ts | 32 +++ .../get-cache-workflow-run-count-key.util.ts | 3 - ...kflow-run-queue-throttle-limit-key.util.ts | 3 - ...ow-run-not-started-count-cache-key.util.ts | 3 + .../workflow-run-queue.module.ts | 19 +- ...ow-handle-staled-runs.workspace-service.ts | 6 +- .../workflow-run-enqueue.workspace-service.ts | 186 ++++++++++++------ .../workflow-run-queue.workspace-service.ts | 135 ------------- .../workflow-throttling.workspace-service.ts | 145 ++++++++++++++ .../workflow-run.workspace-service.ts | 11 +- .../workflow-runner/workflow-runner.module.ts | 2 - .../workflow-runner.workspace-service.ts | 157 ++++++++++----- 25 files changed, 538 insertions(+), 386 deletions(-) delete mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command.ts delete mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/default-workflow-run-queue-throttle-limit.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job.ts delete mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-count-key.util.ts delete mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-queue-throttle-limit-key.util.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util.ts delete mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts diff --git a/packages/twenty-server/src/database/commands/cron-register-all.command.ts b/packages/twenty-server/src/database/commands/cron-register-all.command.ts index 65c496f5b7..2a119b16af 100644 --- a/packages/twenty-server/src/database/commands/cron-register-all.command.ts +++ b/packages/twenty-server/src/database/commands/cron-register-all.command.ts @@ -38,12 +38,14 @@ export class CronRegisterAllCommand extends CommandRunner { private readonly calendarEventsImportCronCommand: CalendarEventsImportCronCommand, private readonly calendarOngoingStaleCronCommand: CalendarOngoingStaleCronCommand, private readonly calendarRelaunchFailedCalendarChannelsCronCommand: CalendarRelaunchFailedCalendarChannelsCronCommand, + private readonly workflowCronTriggerCronCommand: WorkflowCronTriggerCronCommand, - private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand, - private readonly checkPublicDomainsValidRecordsCronCommand: CheckPublicDomainsValidRecordsCronCommand, private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand, private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand, private readonly workflowCleanWorkflowRunsCronCommand: WorkflowCleanWorkflowRunsCronCommand, + + private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand, + private readonly checkPublicDomainsValidRecordsCronCommand: CheckPublicDomainsValidRecordsCronCommand, private readonly cronTriggerCronCommand: CronTriggerCronCommand, private readonly cleanSuspendedWorkspacesCronCommand: CleanSuspendedWorkspacesCronCommand, private readonly cleanOnboardingWorkspacesCronCommand: CleanOnboardingWorkspacesCronCommand, diff --git a/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts b/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts index 64c097708a..570aa51e5b 100644 --- a/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts +++ b/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts @@ -20,7 +20,7 @@ export enum MetricsKeys { WorkflowRunCompleted = 'workflow-run/completed', WorkflowRunFailed = 'workflow-run/failed', WorkflowRunStopped = 'workflow-run/stopped', - WorkflowRunFailedThrottled = 'workflow-run/failed/throttled', + WorkflowRunThrottled = 'workflow-run/throttled', WorkflowRunFailedToEnqueue = 'workflow-run/failed/to-enqueue', AIToolExecutionFailed = 'ai-tool-execution/failed', AIToolExecutionSucceeded = 'ai-tool-execution/succeeded', diff --git a/packages/twenty-server/src/engine/core-modules/throttler/throttler.service.ts b/packages/twenty-server/src/engine/core-modules/throttler/throttler.service.ts index ae33b77bf1..3708d5616a 100644 --- a/packages/twenty-server/src/engine/core-modules/throttler/throttler.service.ts +++ b/packages/twenty-server/src/engine/core-modules/throttler/throttler.service.ts @@ -20,18 +20,14 @@ export class ThrottlerService { tokensToConsume: number, maxTokens: number, timeWindow: number, - ): Promise { + ): Promise { const now = Date.now(); - const refillRate = maxTokens / timeWindow; - - const { tokens, lastRefillAt } = (await this.cacheStorage.get<{ - tokens: number; - lastRefillAt: number; - }>(key)) || { tokens: maxTokens, lastRefillAt: now }; - - const refillAmount = Math.floor((now - lastRefillAt) * refillRate); - - const availableTokens = Math.min(tokens + refillAmount, maxTokens); + const availableTokens = await this.getAvailableTokensCount( + key, + maxTokens, + timeWindow, + now, + ); if (availableTokens < tokensToConsume) { throw new ThrottlerException( @@ -48,5 +44,49 @@ export class ThrottlerService { }, timeWindow * 2, ); + + return availableTokens - tokensToConsume; + } + + async consumeTokens( + key: string, + tokensToConsume: number, + maxTokens: number, + timeWindow: number, + ) { + const now = Date.now(); + const availableTokens = await this.getAvailableTokensCount( + key, + maxTokens, + timeWindow, + now, + ); + + await this.cacheStorage.set( + key, + { + tokens: availableTokens - tokensToConsume, + lastRefillAt: now, + }, + timeWindow * 2, + ); + } + + async getAvailableTokensCount( + key: string, + maxTokens: number, + timeWindow: number, + now = Date.now(), + ): Promise { + const refillRate = maxTokens / timeWindow; + + const { tokens, lastRefillAt } = (await this.cacheStorage.get<{ + tokens: number; + lastRefillAt: number; + }>(key)) || { tokens: maxTokens, lastRefillAt: now }; + + const refillAmount = Math.floor((now - lastRefillAt) * refillRate); + + return Math.min(tokens + refillAmount, maxTokens); } } diff --git a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts index d72b44186a..daa2d370a4 100644 --- a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts +++ b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts @@ -1219,19 +1219,39 @@ export class ConfigVariables { @ConfigVariablesMetadata({ group: ConfigVariablesGroup.RATE_LIMITING, - description: 'Throttle limit for workflow execution', + description: + 'Throttle limit for workflow execution. Remaining will not be enqueued immediately.', type: ConfigVariableType.NUMBER, }) @CastToPositiveNumber() - WORKFLOW_EXEC_THROTTLE_LIMIT = 100; + WORKFLOW_EXEC_SOFT_THROTTLE_LIMIT = 100; @ConfigVariablesMetadata({ group: ConfigVariablesGroup.RATE_LIMITING, - description: 'Time-to-live for workflow execution throttle in milliseconds', + description: + 'Time-to-live for workflow execution throttle in milliseconds. Remaining will not be enqueued immediately.', type: ConfigVariableType.NUMBER, }) @CastToPositiveNumber() - WORKFLOW_EXEC_THROTTLE_TTL = 60_000; + WORKFLOW_EXEC_SOFT_THROTTLE_TTL = 60_000; + + @ConfigVariablesMetadata({ + group: ConfigVariablesGroup.RATE_LIMITING, + description: + 'Throttle limit for workflow execution. Remaining will be marked as failed.', + type: ConfigVariableType.NUMBER, + }) + @CastToPositiveNumber() + WORKFLOW_EXEC_HARD_THROTTLE_LIMIT = 5000; + + @ConfigVariablesMetadata({ + group: ConfigVariablesGroup.RATE_LIMITING, + description: + 'Time-to-live for workflow execution throttle in milliseconds. Remaining will be marked as failed.', + type: ConfigVariableType.NUMBER, + }) + @CastToPositiveNumber() + WORKFLOW_EXEC_HARD_THROTTLE_TTL = 3_600_000; // 1 hour; @ConfigVariablesMetadata({ group: ConfigVariablesGroup.CAPTCHA_CONFIG, diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts index 66d7803913..512d2f6bee 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts @@ -17,7 +17,6 @@ import { } from 'src/modules/workflow/workflow-runner/exceptions/workflow-run.exception'; import { RunWorkflowJob } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; @Processor({ @@ -29,7 +28,6 @@ export class ResumeDelayedWorkflowJob { @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService, - private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, ) {} @Process(RESUME_DELAYED_WORKFLOW_JOB_NAME) @@ -89,10 +87,6 @@ export class ResumeDelayedWorkflowJob { lastExecutedStepId: stepId, }, ); - - await this.workflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount( - workspaceId, - ); } catch (error) { await this.workflowRunWorkspaceService.endWorkflowRun({ workflowRunId, diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts index 9a90f350a6..b658f96fc4 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts @@ -16,24 +16,22 @@ import { IteratorActionModule } from 'src/modules/workflow/workflow-executor/wor import { RecordCRUDActionModule } from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/record-crud-action.module'; import { ToolExecutorWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/tool-executor-workflow-action'; import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service'; -import { WorkflowRunQueueModule } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module'; import { WorkflowRunModule } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.module'; @Module({ imports: [ WorkflowCommonModule, + WorkflowRunModule, CodeActionModule, DelayActionModule, RecordCRUDActionModule, FormActionModule, - WorkflowRunModule, BillingModule, FilterActionModule, IteratorActionModule, AiAgentActionModule, EmptyActionModule, FeatureFlagModule, - WorkflowRunQueueModule, ToolModule, ], providers: [ diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts index e063942864..3c8935fbc5 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts @@ -15,7 +15,6 @@ import { WorkflowActionType, } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type'; import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; jest.mock( @@ -61,10 +60,6 @@ describe('WorkflowExecutorWorkspaceService', () => { add: jest.fn(), }; - const mockWorkflowRunQueueWorkspaceService = { - increaseWorkflowRunQueuedCount: jest.fn(), - }; - beforeEach(async () => { jest.clearAllMocks(); @@ -93,10 +88,6 @@ describe('WorkflowExecutorWorkspaceService', () => { provide: `MESSAGE_QUEUE_${MessageQueue.workflowQueue}`, useValue: mockMessageQueueService, }, - { - provide: WorkflowRunQueueWorkspaceService, - useValue: mockWorkflowRunQueueWorkspaceService, - }, ], }).compile(); @@ -383,10 +374,6 @@ describe('WorkflowExecutorWorkspaceService', () => { }, ); - expect( - mockWorkflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount, - ).toHaveBeenCalledWith(mockWorkspaceId); - // Should not execute the next step (step-2) in the same job expect(workflowActionFactory.get).toHaveBeenCalledTimes(1); expect(workflowActionFactory.get).toHaveBeenCalledWith( diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts index d0660b4a4f..51f63c7f64 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts @@ -36,7 +36,6 @@ import { WorkflowIteratorResult } from 'src/modules/workflow/workflow-executor/w import { WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type'; import { RUN_WORKFLOW_JOB_NAME } from 'src/modules/workflow/workflow-runner/constants/run-workflow-job-name'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; const MAX_EXECUTED_STEPS_COUNT = 20; @@ -50,7 +49,6 @@ export class WorkflowExecutorWorkspaceService { private readonly billingService: BillingService, @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, - private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, ) {} async executeFromSteps({ @@ -418,8 +416,5 @@ export class WorkflowExecutorWorkspaceService { lastExecutedStepId, }, ); - await this.workflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount( - workspaceId, - ); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts index 777e3f6ceb..d0c414c030 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts @@ -7,8 +7,6 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; -import { ThrottlerService } from 'src/engine/core-modules/throttler/throttler.service'; -import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service'; import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service'; @@ -18,7 +16,6 @@ import { WorkflowRunExceptionCode, } from 'src/modules/workflow/workflow-runner/exceptions/workflow-run.exception'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; import { WorkflowTriggerType } from 'src/modules/workflow/workflow-trigger/types/workflow-trigger.type'; @@ -28,10 +25,7 @@ export class RunWorkflowJob { private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService, private readonly workflowExecutorWorkspaceService: WorkflowExecutorWorkspaceService, private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService, - private readonly throttlerService: ThrottlerService, - private readonly twentyConfigService: TwentyConfigService, private readonly metricsService: MetricsService, - private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, ) {} @Process(RUN_WORKFLOW_JOB_NAME) @@ -60,10 +54,6 @@ export class RunWorkflowJob { status: WorkflowRunStatus.FAILED, error: error.message, }); - } finally { - await this.workflowRunQueueWorkspaceService.decreaseWorkflowRunQueuedCount( - workspaceId, - ); } } @@ -93,18 +83,16 @@ export class RunWorkflowJob { ); } - await this.throttleExecution(workflowVersion.workflowId); + await this.workflowRunWorkspaceService.startWorkflowRun({ + workflowRunId, + workspaceId, + }); await this.incrementTriggerMetrics({ workflowRunId, triggerType: workflowVersion.trigger.type, }); - await this.workflowRunWorkspaceService.startWorkflowRun({ - workflowRunId, - workspaceId, - }); - const stepIds = workflowVersion.trigger.nextStepIds ?? []; await this.workflowExecutorWorkspaceService.executeFromSteps({ @@ -170,27 +158,6 @@ export class RunWorkflowJob { }); } - private async throttleExecution(workflowId: string) { - try { - await this.throttlerService.tokenBucketThrottleOrThrow( - `${workflowId}-workflow-execution`, - 1, - this.twentyConfigService.get('WORKFLOW_EXEC_THROTTLE_LIMIT'), - this.twentyConfigService.get('WORKFLOW_EXEC_THROTTLE_TTL'), - ); - } catch { - await this.metricsService.incrementCounter({ - key: MetricsKeys.WorkflowRunFailedThrottled, - eventId: workflowId, - }); - - throw new WorkflowRunException( - 'Workflow execution rate limit exceeded', - WorkflowRunExceptionCode.WORKFLOW_RUN_LIMIT_REACHED, - ); - } - } - private async incrementTriggerMetrics({ workflowRunId, triggerType, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command.ts deleted file mode 100644 index b0400af45f..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command.ts +++ /dev/null @@ -1,39 +0,0 @@ -import { Command, CommandRunner, Option } from 'nest-commander'; - -import { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service'; - -type WorkflowRunEnqueueCommandOptions = { - workspaceIds: string[]; -}; - -@Command({ - name: 'workflow:run:enqueue', - description: 'Enqueues not started workflow runs', -}) -export class WorkflowRunEnqueueCommand extends CommandRunner { - constructor( - private readonly workflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService, - ) { - super(); - } - - @Option({ - flags: '-w, --workspace-ids [workspace_ids]', - description: 'comma separated workspace ids - mandatory', - required: true, - }) - parseWorkspaceIds(val: string): string[] { - return val.split(','); - } - - async run( - _passedParam: string[], - options: WorkflowRunEnqueueCommandOptions, - ): Promise { - const { workspaceIds } = options; - - await this.workflowRunEnqueueWorkspaceService.enqueueRuns({ - workspaceIds, - }); - } -} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/default-workflow-run-queue-throttle-limit.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/default-workflow-run-queue-throttle-limit.ts deleted file mode 100644 index 86c818e318..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/default-workflow-run-queue-throttle-limit.ts +++ /dev/null @@ -1 +0,0 @@ -export const DEFAULT_WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts index eb0726ff8b..075a7c13ae 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts @@ -5,7 +5,7 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; import { WORKFLOW_RUN_ENQUEUE_CRON_PATTERN, - WorkflowRunEnqueueJob, + WorkflowRunEnqueueCronJob, } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job'; @Command({ @@ -22,7 +22,7 @@ export class WorkflowRunEnqueueCronCommand extends CommandRunner { async run(): Promise { await this.messageQueueService.addCron({ - jobName: WorkflowRunEnqueueJob.name, + jobName: WorkflowRunEnqueueCronJob.name, data: undefined, options: { repeat: { diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts index 8cb8965d1c..d155628df0 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts @@ -13,16 +13,16 @@ import { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflo export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '*/5 * * * *'; @Processor(MessageQueue.cronQueue) -export class WorkflowRunEnqueueJob { +export class WorkflowRunEnqueueCronJob { constructor( @InjectRepository(WorkspaceEntity) private readonly workspaceRepository: Repository, - private readonly workflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService, + private readonly WorkflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService, ) {} - @Process(WorkflowRunEnqueueJob.name) + @Process(WorkflowRunEnqueueCronJob.name) @SentryCronMonitor( - WorkflowRunEnqueueJob.name, + WorkflowRunEnqueueCronJob.name, WORKFLOW_RUN_ENQUEUE_CRON_PATTERN, ) async handle() { @@ -32,8 +32,9 @@ export class WorkflowRunEnqueueJob { }, }); - await this.workflowRunEnqueueWorkspaceService.enqueueRuns({ + await this.WorkflowRunEnqueueWorkspaceService.enqueueRuns({ workspaceIds: activeWorkspaces.map((workspace) => workspace.id), + isCacheMode: false, }); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job.ts new file mode 100644 index 0000000000..6ce807b8cd --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job.ts @@ -0,0 +1,32 @@ +import { Scope } from '@nestjs/common'; + +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 { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service'; + +export type WorkflowRunEnqueueJobData = { + workspaceId: string; + isCacheMode: boolean; + priorityWorkflowRunId?: string; +}; + +@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST }) +export class WorkflowRunEnqueueJob { + constructor( + private readonly WorkflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService, + ) {} + + @Process(WorkflowRunEnqueueJob.name) + async handle({ + workspaceId, + isCacheMode, + priorityWorkflowRunId, + }: WorkflowRunEnqueueJobData): Promise { + await this.WorkflowRunEnqueueWorkspaceService.enqueueRunsForWorkspace({ + workspaceId, + priorityWorkflowRunId, + isCacheMode, + }); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-count-key.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-count-key.util.ts deleted file mode 100644 index 8dec189c8c..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-count-key.util.ts +++ /dev/null @@ -1,3 +0,0 @@ -export const getWorkflowRunQueuedCountCacheKey = ( - workspaceId: string, -): string => `workflow-run-queued-count:${workspaceId}`; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-queue-throttle-limit-key.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-queue-throttle-limit-key.util.ts deleted file mode 100644 index 1556260922..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-queue-throttle-limit-key.util.ts +++ /dev/null @@ -1,3 +0,0 @@ -export const getWorkflowRunQueueThrottleLimitKey = ( - workspaceId: string, -): string => `workflow-run-queue-throttle-limit:${workspaceId}`; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util.ts new file mode 100644 index 0000000000..952b422089 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util.ts @@ -0,0 +1,3 @@ +export const getWorkflowRunNotStartedCountCacheKey = ( + workspaceId: string, +): string => `workflow-run-not-started-count:${workspaceId}`; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts index acf7dbfab7..25c68baf68 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts @@ -4,19 +4,20 @@ import { TypeOrmModule } from '@nestjs/typeorm'; import { CacheStorageModule } from 'src/engine/core-modules/cache-storage/cache-storage.module'; import { MessageQueueModule } from 'src/engine/core-modules/message-queue/message-queue.module'; import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module'; +import { ThrottlerModule } from 'src/engine/core-modules/throttler/throttler.module'; import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; import { WorkflowHandleStaledRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command'; -import { WorkflowRunEnqueueCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-run-enqueue.command'; import { WorkflowCleanWorkflowRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command'; import { WorkflowHandleStaledRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command'; import { WorkflowRunEnqueueCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command'; import { WorkflowCleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job'; import { WorkflowHandleStaledRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job'; -import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job'; +import { WorkflowRunEnqueueCronJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job'; +import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job'; import { WorkflowHandleStaledRunsWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service'; import { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; +import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service'; @Module({ imports: [ @@ -25,12 +26,13 @@ import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow- MessageQueueModule, WorkspaceDataSourceModule, MetricsModule, + ThrottlerModule, ], providers: [ - WorkflowRunQueueWorkspaceService, - WorkflowRunEnqueueWorkspaceService, + WorkflowThrottlingWorkspaceService, + WorkflowRunEnqueueCronJob, WorkflowRunEnqueueCronCommand, - WorkflowRunEnqueueCommand, + WorkflowRunEnqueueWorkspaceService, WorkflowRunEnqueueJob, WorkflowHandleStaledRunsWorkspaceService, WorkflowHandleStaledRunsCronCommand, @@ -40,9 +42,10 @@ import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow- WorkflowCleanWorkflowRunsCronCommand, ], exports: [ - WorkflowRunQueueWorkspaceService, + WorkflowThrottlingWorkspaceService, + WorkflowRunEnqueueJob, + WorkflowRunEnqueueCronJob, WorkflowRunEnqueueCronCommand, - WorkflowRunEnqueueCommand, WorkflowHandleStaledRunsCronCommand, WorkflowHandleStaledRunsCommand, WorkflowCleanWorkflowRunsCronCommand, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts index 7ee7b7292d..aa539e30c0 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts @@ -7,7 +7,7 @@ import { WorkflowRunStatus, WorkflowRunWorkspaceEntity, } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; +import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service'; @Injectable() export class WorkflowHandleStaledRunsWorkspaceService { @@ -16,7 +16,7 @@ export class WorkflowHandleStaledRunsWorkspaceService { ); constructor( private readonly twentyORMGlobalManager: TwentyORMGlobalManager, - private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, + private readonly workflowThrottlingWorkspaceService: WorkflowThrottlingWorkspaceService, ) {} async handleStaledRuns({ workspaceIds }: { workspaceIds: string[] }) { @@ -50,7 +50,7 @@ export class WorkflowHandleStaledRunsWorkspaceService { }, ); - await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( + await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount( workspaceId, ); } catch (error) { diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts index 9b28d9819d..6472b3e1cf 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts @@ -1,5 +1,8 @@ import { Injectable, Logger } from '@nestjs/common'; +import { isDefined } from 'twenty-shared/utils'; +import { Not } from 'typeorm'; + import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; @@ -12,45 +15,100 @@ import { } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { RunWorkflowJob } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; +import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service'; @Injectable() export class WorkflowRunEnqueueWorkspaceService { private readonly logger = new Logger(WorkflowRunEnqueueWorkspaceService.name); constructor( - private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, + private readonly workflowThrottlingWorkspaceService: WorkflowThrottlingWorkspaceService, private readonly twentyORMGlobalManager: TwentyORMGlobalManager, @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, private readonly metricsService: MetricsService, ) {} - async enqueueRuns({ workspaceIds }: { workspaceIds: string[] }) { + async enqueueRuns({ + workspaceIds, + isCacheMode, + }: { + workspaceIds: string[]; + isCacheMode: boolean; + }) { for (const workspaceId of workspaceIds) { - try { - const workflowRunRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - workspaceId, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); + await this.enqueueRunsForWorkspace({ workspaceId, isCacheMode }); + } + } - const remainingWorkflowRunToEnqueueCount = - await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromDatabase( + async enqueueRunsForWorkspace({ + workspaceId, + priorityWorkflowRunId, + isCacheMode, + }: { + workspaceId: string; + priorityWorkflowRunId?: string; + isCacheMode: boolean; + }) { + try { + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const notStartedRunsCount = isCacheMode + ? await this.workflowThrottlingWorkspaceService.getNotStartedRunsCountFromCache( + workspaceId, + ) + : await this.workflowThrottlingWorkspaceService.getNotStartedRunsCountFromDatabase( workspaceId, ); - if (remainingWorkflowRunToEnqueueCount <= 0) { - await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( + if (notStartedRunsCount <= 0) { + if (!isCacheMode) { + await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount( workspaceId, ); - - continue; } - const workflowRunsToEnqueue = await workflowRunRepository.find({ + return; + } + + let remainingWorkflowRunToEnqueueCount = + await this.workflowThrottlingWorkspaceService.getRemainingRunsToEnqueueCount( + workspaceId, + ); + + const workflowRunIdsToEnqueue: string[] = []; + + if (isDefined(priorityWorkflowRunId)) { + const priorityRun = await workflowRunRepository.findOne({ + where: { + id: priorityWorkflowRunId, + status: WorkflowRunStatus.NOT_STARTED, + }, + select: { + id: true, + }, + }); + + if (isDefined(priorityRun)) { + workflowRunIdsToEnqueue.push(priorityRun.id); + remainingWorkflowRunToEnqueueCount--; + } + } + + if (remainingWorkflowRunToEnqueueCount > 0) { + const additionalRunsToEnqueue = await workflowRunRepository.find({ where: { status: WorkflowRunStatus.NOT_STARTED, + ...(workflowRunIdsToEnqueue.length > 0 + ? { id: Not(workflowRunIdsToEnqueue[0]) } + : {}), + }, + select: { + id: true, }, order: { createdAt: 'ASC', @@ -58,47 +116,63 @@ export class WorkflowRunEnqueueWorkspaceService { take: remainingWorkflowRunToEnqueueCount, }); - if (workflowRunsToEnqueue.length <= 0) { - await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( - workspaceId, - ); - - continue; - } - - const workflowRunIds = workflowRunsToEnqueue.map( - (workflowRun: WorkflowRunWorkspaceEntity) => workflowRun.id, - ); - - await workflowRunRepository.update(workflowRunIds, { - enqueuedAt: new Date().toISOString(), - status: WorkflowRunStatus.ENQUEUED, - }); - - for (const workflowRunId of workflowRunIds) { - await this.messageQueueService.add( - RunWorkflowJob.name, - { - workflowRunId, - workspaceId, - }, - ); - } - - await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( - workspaceId, - ); - } catch (error) { - this.metricsService.incrementCounter({ - key: MetricsKeys.WorkflowRunFailedToEnqueue, - eventId: workspaceId, - }); - - this.logger.error( - `Failed to enqueue workflow runs for workspace: ${workspaceId}`, - error, + workflowRunIdsToEnqueue.push( + ...additionalRunsToEnqueue.map( + (workflowRun: WorkflowRunWorkspaceEntity) => workflowRun.id, + ), ); } + + if (workflowRunIdsToEnqueue.length <= 0) { + if (!isCacheMode) { + await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount( + workspaceId, + ); + } + + return; + } + + await workflowRunRepository.update(workflowRunIdsToEnqueue, { + enqueuedAt: new Date().toISOString(), + status: WorkflowRunStatus.ENQUEUED, + }); + + await this.workflowThrottlingWorkspaceService.consumeRemainingRunsToEnqueueCount( + workspaceId, + workflowRunIdsToEnqueue.length, + ); + + for (const workflowRunId of workflowRunIdsToEnqueue) { + await this.messageQueueService.add( + RunWorkflowJob.name, + { + workflowRunId, + workspaceId, + }, + ); + } + + if (isCacheMode) { + await this.workflowThrottlingWorkspaceService.decreaseWorkflowRunNotStartedCount( + workspaceId, + workflowRunIdsToEnqueue.length, + ); + } else { + await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount( + workspaceId, + ); + } + } catch (error) { + this.metricsService.incrementCounter({ + key: MetricsKeys.WorkflowRunFailedToEnqueue, + eventId: workspaceId, + }); + + this.logger.error( + `Failed to enqueue workflow runs for workspace: ${workspaceId}`, + error, + ); } } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts deleted file mode 100644 index dd723f8107..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts +++ /dev/null @@ -1,135 +0,0 @@ -import { Injectable } from '@nestjs/common'; - -import { In } from 'typeorm'; - -import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; -import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; -import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; -import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { - WorkflowRunStatus, - WorkflowRunWorkspaceEntity, -} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; -import { DEFAULT_WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/default-workflow-run-queue-throttle-limit'; -import { getWorkflowRunQueuedCountCacheKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-count-key.util'; -import { getWorkflowRunQueueThrottleLimitKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-queue-throttle-limit-key.util'; - -@Injectable() -export class WorkflowRunQueueWorkspaceService { - constructor( - @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) - private readonly cacheStorage: CacheStorageService, - private readonly twentyORMGlobalManager: TwentyORMGlobalManager, - ) {} - - async increaseWorkflowRunQueuedCount( - workspaceId: string, - newlyEnqueuedCount = 1, - ): Promise { - const currentCount = - await this.getCurrentWorkflowRunQueuedCount(workspaceId); - - await this.cacheStorage.set( - getWorkflowRunQueuedCountCacheKey(workspaceId), - currentCount + newlyEnqueuedCount, - ); - } - - async decreaseWorkflowRunQueuedCount( - workspaceId: string, - removedFromQueueCount = 1, - ): Promise { - const currentCount = - await this.getCurrentWorkflowRunQueuedCount(workspaceId); - - await this.cacheStorage.set( - getWorkflowRunQueuedCountCacheKey(workspaceId), - currentCount - removedFromQueueCount, - ); - } - - async recomputeWorkflowRunQueuedCount(workspaceId: string): Promise { - const workflowRunRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - workspaceId, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); - - const currentlyEnqueuedWorkflowRunCount = await workflowRunRepository.count( - { - where: { - status: In([WorkflowRunStatus.ENQUEUED, WorkflowRunStatus.RUNNING]), - }, - }, - ); - - await this.setWorkflowRunQueuedCount( - workspaceId, - currentlyEnqueuedWorkflowRunCount, - ); - } - - async getRemainingRunsToEnqueueCountFromCache( - workspaceId: string, - ): Promise { - const currentCount = - await this.getCurrentWorkflowRunQueuedCount(workspaceId); - const throttleLimit = - await this.getWorkflowRunQueueThrottleLimit(workspaceId); - - return throttleLimit - currentCount; - } - - async getRemainingRunsToEnqueueCountFromDatabase( - workspaceId: string, - ): Promise { - const workflowRunRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - workspaceId, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); - - const currentCount = await workflowRunRepository.count({ - where: { - status: In([WorkflowRunStatus.ENQUEUED, WorkflowRunStatus.RUNNING]), - }, - }); - - const throttleLimit = - await this.getWorkflowRunQueueThrottleLimit(workspaceId); - - return throttleLimit - currentCount; - } - - private async setWorkflowRunQueuedCount( - workspaceId: string, - count: number, - ): Promise { - await this.cacheStorage.set( - getWorkflowRunQueuedCountCacheKey(workspaceId), - count, - ); - } - - private async getCurrentWorkflowRunQueuedCount( - workspaceId: string, - ): Promise { - const key = getWorkflowRunQueuedCountCacheKey(workspaceId); - - const currentCount = (await this.cacheStorage.get(key)) ?? 0; - - return Math.max(0, currentCount); - } - - private async getWorkflowRunQueueThrottleLimit( - workspaceId: string, - ): Promise { - const key = getWorkflowRunQueueThrottleLimitKey(workspaceId); - - const throttleLimit = (await this.cacheStorage.get(key)) ?? 0; - - return Math.max(DEFAULT_WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT, throttleLimit); - } -} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts new file mode 100644 index 0000000000..90323cc70c --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service.ts @@ -0,0 +1,145 @@ +import { Injectable } from '@nestjs/common'; + +import { In } from 'typeorm'; + +import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; +import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; +import { ThrottlerService } from 'src/engine/core-modules/throttler/throttler.service'; +import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; +import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; +import { + WorkflowRunStatus, + WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { getWorkflowRunNotStartedCountCacheKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util'; + +@Injectable() +export class WorkflowThrottlingWorkspaceService { + constructor( + @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) + private readonly cacheStorage: CacheStorageService, + private readonly twentyORMGlobalManager: TwentyORMGlobalManager, + private readonly throttlerService: ThrottlerService, + private readonly twentyConfigService: TwentyConfigService, + ) {} + + async getRemainingRunsToEnqueueCount(workspaceId: string) { + return this.throttlerService.getAvailableTokensCount( + `${workspaceId}-workflow-execution-soft-throttle`, + this.twentyConfigService.get('WORKFLOW_EXEC_SOFT_THROTTLE_LIMIT'), + this.twentyConfigService.get('WORKFLOW_EXEC_SOFT_THROTTLE_TTL'), + ); + } + + async consumeRemainingRunsToEnqueueCount( + workspaceId: string, + runsToConsume: number, + ) { + await this.throttlerService.consumeTokens( + `${workspaceId}-workflow-execution-soft-throttle`, + runsToConsume, + this.twentyConfigService.get('WORKFLOW_EXEC_SOFT_THROTTLE_LIMIT'), + this.twentyConfigService.get('WORKFLOW_EXEC_SOFT_THROTTLE_TTL'), + ); + } + + async throttleOrThrowIfHardLimitReached(workspaceId: string) { + await this.throttlerService.tokenBucketThrottleOrThrow( + `${workspaceId}-workflow-execution-hard-throttle`, + 1, + this.twentyConfigService.get('WORKFLOW_EXEC_HARD_THROTTLE_LIMIT'), + this.twentyConfigService.get('WORKFLOW_EXEC_HARD_THROTTLE_TTL'), + ); + } + + async increaseWorkflowRunNotStartedCount( + workspaceId: string, + newlyEnqueuedCount = 1, + ): Promise { + const currentCount = + await this.getCurrentWorkflowRunNotStartedCount(workspaceId); + + await this.cacheStorage.set( + getWorkflowRunNotStartedCountCacheKey(workspaceId), + currentCount + newlyEnqueuedCount, + ); + } + + async decreaseWorkflowRunNotStartedCount( + workspaceId: string, + removedFromQueueCount = 1, + ): Promise { + const currentCount = + await this.getCurrentWorkflowRunNotStartedCount(workspaceId); + + await this.cacheStorage.set( + getWorkflowRunNotStartedCountCacheKey(workspaceId), + currentCount - removedFromQueueCount, + ); + } + + async recomputeWorkflowRunNotStartedCount( + workspaceId: string, + ): Promise { + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const currentlyNotStartedWorkflowRunCount = + await workflowRunRepository.count({ + where: { + status: In([WorkflowRunStatus.NOT_STARTED]), + }, + }); + + await this.setWorkflowRunNotStartedCount( + workspaceId, + currentlyNotStartedWorkflowRunCount, + ); + } + + async getNotStartedRunsCountFromCache(workspaceId: string): Promise { + return this.getCurrentWorkflowRunNotStartedCount(workspaceId); + } + + async getNotStartedRunsCountFromDatabase( + workspaceId: string, + ): Promise { + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + return workflowRunRepository.count({ + where: { + status: In([WorkflowRunStatus.NOT_STARTED]), + }, + }); + } + + private async setWorkflowRunNotStartedCount( + workspaceId: string, + count: number, + ): Promise { + await this.cacheStorage.set( + getWorkflowRunNotStartedCountCacheKey(workspaceId), + count, + ); + } + + private async getCurrentWorkflowRunNotStartedCount( + workspaceId: string, + ): Promise { + const key = getWorkflowRunNotStartedCountCacheKey(workspaceId); + + const currentCount = (await this.cacheStorage.get(key)) ?? 0; + + return Math.max(0, currentCount); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts index cb92d0d308..ebcbaabeaa 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts @@ -41,12 +41,17 @@ export class WorkflowRunWorkspaceService { workflowRunId, status, triggerPayload, + error, }: { workflowVersionId: string; createdBy: ActorMetadata; - status: WorkflowRunStatus.NOT_STARTED | WorkflowRunStatus.ENQUEUED; + status: + | WorkflowRunStatus.NOT_STARTED + | WorkflowRunStatus.ENQUEUED + | WorkflowRunStatus.FAILED; triggerPayload: object; workflowRunId?: string; + error?: string; }) { const workspaceId = this.scopedWorkspaceContextFactory.create()?.workspaceId; @@ -100,7 +105,7 @@ export class WorkflowRunWorkspaceService { workspaceId, }); - const initState = this.getInitState(workflowVersion, triggerPayload); + const initState = this.getInitState(workflowVersion, triggerPayload, error); const lastWorkflowRun = await workflowRunRepository.findOne({ where: { @@ -415,6 +420,7 @@ export class WorkflowRunWorkspaceService { private getInitState( workflowVersion: WorkflowVersionWorkspaceEntity, triggerPayload: object, + error?: string, ): WorkflowRunState | undefined { if ( !isDefined(workflowVersion.trigger) || @@ -437,6 +443,7 @@ export class WorkflowRunWorkspaceService { ]), ), }, + workflowRunError: error, }; } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-runner.module.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-runner.module.ts index 7d23f4292c..f75fe8df06 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-runner.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-runner.module.ts @@ -2,7 +2,6 @@ import { Module } from '@nestjs/common'; import { BillingModule } from 'src/engine/core-modules/billing/billing.module'; import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module'; -import { ThrottlerModule } from 'src/engine/core-modules/throttler/throttler.module'; import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module'; import { WorkflowVersionStepModule } from 'src/modules/workflow/workflow-builder/workflow-version-step/workflow-version-step.module'; import { WorkflowExecutorModule } from 'src/modules/workflow/workflow-executor/workflow-executor.module'; @@ -15,7 +14,6 @@ import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-ru imports: [ WorkflowCommonModule, WorkflowExecutorModule, - ThrottlerModule, BillingModule, WorkflowRunModule, MetricsModule, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts index cb9b0f8e1f..d59e2cd4bc 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts @@ -1,14 +1,16 @@ import { Injectable, Logger } from '@nestjs/common'; import { msg } from '@lingui/core/macro'; +import { type ActorMetadata } from 'twenty-shared/types'; import { isDefined } from 'twenty-shared/utils'; import { StepStatus } from 'twenty-shared/workflow'; -import { type ActorMetadata } from 'twenty-shared/types'; import { BillingUsageService } from 'src/engine/core-modules/billing/services/billing-usage.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; +import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; import { WorkflowVersionStepException, WorkflowVersionStepExceptionCode, @@ -24,7 +26,11 @@ import { } from 'src/modules/workflow/workflow-runner/exceptions/workflow-run.exception'; import { RunWorkflowJob } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; +import { + WorkflowRunEnqueueJob, + type WorkflowRunEnqueueJobData, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job'; +import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; import { WorkflowTriggerType } from 'src/modules/workflow/workflow-trigger/types/workflow-trigger.type'; @@ -34,11 +40,12 @@ export class WorkflowRunnerWorkspaceService { constructor( private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService, private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService, - private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, private readonly billingUsageService: BillingUsageService, private readonly workflowVersionStepOperationsWorkspaceService: WorkflowVersionStepOperationsWorkspaceService, + private readonly workflowThrottlingWorkspaceService: WorkflowThrottlingWorkspaceService, + private readonly metricsService: MetricsService, ) {} async run({ @@ -69,34 +76,28 @@ export class WorkflowRunnerWorkspaceService { workflowVersionId, }); - const remainingRunsToEnqueueCount = - await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromCache( - workspaceId, - ); - - const isQueueLimitReached = remainingRunsToEnqueueCount <= 0; - const isManualTrigger = workflowVersion.trigger?.type === WorkflowTriggerType.MANUAL; - const shouldEnqueueWorkflowRun = isManualTrigger || !isQueueLimitReached; + const isHardThrottled = await this.checkHardThrottleLimit(workspaceId); - const workflowRunId = - await this.workflowRunWorkspaceService.createWorkflowRun({ + if (isHardThrottled) { + return this.createFailedWorkflowRun({ workflowVersionId, - workflowRunId: initialWorkflowRunId, - createdBy: source, - status: shouldEnqueueWorkflowRun - ? WorkflowRunStatus.ENQUEUED - : WorkflowRunStatus.NOT_STARTED, - triggerPayload: payload, + initialWorkflowRunId, + source, + payload, }); - - if (shouldEnqueueWorkflowRun) { - await this.enqueueWorkflowRun(workspaceId, workflowRunId); } - return { workflowRunId }; + return this.createNotStartedWorkflowRunAndTriggerEnqueue({ + workspaceId, + workflowVersionId, + initialWorkflowRunId, + source, + payload, + isManualTrigger, + }); } async resume({ @@ -108,10 +109,13 @@ export class WorkflowRunnerWorkspaceService { workflowRunId: string; lastExecutedStepId: string; }) { - await this.enqueueWorkflowRun( - workspaceId, - workflowRunId, - lastExecutedStepId, + await this.messageQueueService.add( + RunWorkflowJob.name, + { + workspaceId, + workflowRunId, + lastExecutedStepId, + }, ); } @@ -179,24 +183,6 @@ export class WorkflowRunnerWorkspaceService { }); } - async enqueueWorkflowRun( - workspaceId: string, - workflowRunId: string, - lastExecutedStepId?: string, - ) { - await this.messageQueueService.add( - RunWorkflowJob.name, - { - workspaceId, - workflowRunId, - lastExecutedStepId, - }, - ); - await this.workflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount( - workspaceId, - ); - } - async stopWorkflowRun(workspaceId: string, workflowRunId: string) { const workflowRun = await this.workflowRunWorkspaceService.getWorkflowRunOrFail({ @@ -244,4 +230,85 @@ export class WorkflowRunnerWorkspaceService { status: newStatus, }; } + + private async checkHardThrottleLimit(workspaceId: string): Promise { + try { + await this.workflowThrottlingWorkspaceService.throttleOrThrowIfHardLimitReached( + workspaceId, + ); + + return false; + } catch { + this.metricsService.incrementCounter({ + key: MetricsKeys.WorkflowRunThrottled, + eventId: workspaceId, + }); + + return true; + } + } + + private async createFailedWorkflowRun({ + workflowVersionId, + initialWorkflowRunId, + source, + payload, + }: { + workflowVersionId: string; + initialWorkflowRunId?: string; + source: ActorMetadata; + payload: object; + }) { + const workflowRunId = + await this.workflowRunWorkspaceService.createWorkflowRun({ + workflowVersionId, + workflowRunId: initialWorkflowRunId, + createdBy: source, + status: WorkflowRunStatus.FAILED, + triggerPayload: payload, + error: 'Throttle limit reached', + }); + + return { workflowRunId }; + } + + private async createNotStartedWorkflowRunAndTriggerEnqueue({ + workspaceId, + workflowVersionId, + initialWorkflowRunId, + source, + payload, + isManualTrigger, + }: { + workspaceId: string; + workflowVersionId: string; + initialWorkflowRunId?: string; + source: ActorMetadata; + payload: object; + isManualTrigger: boolean; + }) { + const workflowRunId = + await this.workflowRunWorkspaceService.createWorkflowRun({ + workflowVersionId, + workflowRunId: initialWorkflowRunId, + createdBy: source, + status: WorkflowRunStatus.NOT_STARTED, + triggerPayload: payload, + }); + + await this.workflowThrottlingWorkspaceService.increaseWorkflowRunNotStartedCount( + workspaceId, + ); + + await this.messageQueueService.add( + WorkflowRunEnqueueJob.name, + { + workspaceId, + priorityWorkflowRunId: isManualTrigger ? workflowRunId : undefined, + isCacheMode: true, + }, + ); + + return { workflowRunId }; + } }