From 4a94ad10036885609334fc8c895bca40390d9a62 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Thu, 4 Dec 2025 14:27:22 +0100 Subject: [PATCH] Add lock to enqueue workflow job (#16314) - Add lock to run one enqueue job at a time. It will avoid race conditions - Enqueue manual workflows directly, without waiting for the job --- .../services/cache-storage.service.ts | 16 +++++ .../jobs/workflow-run-enqueue.cron.job.ts | 4 +- .../jobs/workflow-run-enqueue.job.ts | 3 - ...ow-run-not-started-count-cache-key.util.ts | 3 - .../workflow-run-enqueue.workspace-service.ts | 41 +++++------- .../workflow-throttling.workspace-service.ts | 64 +++++++++++++------ .../workflow-runner.workspace-service.ts | 51 +++++++++++++-- 7 files changed, 125 insertions(+), 57 deletions(-) delete mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util.ts diff --git a/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts b/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts index 8a1003d5ee..23519d9102 100644 --- a/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts +++ b/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts @@ -208,6 +208,22 @@ export class CacheStorageService { await this.del(key); } + async incrBy(key: string, increment: number): Promise { + if (this.isRedisCache()) { + return (this.cache as RedisCache).store.client.incrBy( + this.getKey(key), + increment, + ); + } + + const current = (await this.get(key)) ?? 0; + const newValue = current + increment; + + await this.set(key, newValue); + + return newValue; + } + private isRedisCache() { // eslint-disable-next-line @typescript-eslint/no-explicit-any return (this.cache.store as any)?.name === 'redis'; 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 d155628df0..e2182ce256 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 @@ -17,7 +17,7 @@ export class WorkflowRunEnqueueCronJob { constructor( @InjectRepository(WorkspaceEntity) private readonly workspaceRepository: Repository, - private readonly WorkflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService, + private readonly workflowRunEnqueueWorkspaceService: WorkflowRunEnqueueWorkspaceService, ) {} @Process(WorkflowRunEnqueueCronJob.name) @@ -32,7 +32,7 @@ export class WorkflowRunEnqueueCronJob { }, }); - 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 index 6ce807b8cd..00dfda63bc 100644 --- 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 @@ -8,7 +8,6 @@ import { WorkflowRunEnqueueWorkspaceService } from 'src/modules/workflow/workflo export type WorkflowRunEnqueueJobData = { workspaceId: string; isCacheMode: boolean; - priorityWorkflowRunId?: string; }; @Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST }) @@ -21,11 +20,9 @@ export class WorkflowRunEnqueueJob { 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-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 deleted file mode 100644 index 952b422089..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util.ts +++ /dev/null @@ -1,3 +0,0 @@ -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/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 6472b3e1cf..aad7114455 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,6 +1,5 @@ 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'; @@ -42,13 +41,20 @@ export class WorkflowRunEnqueueWorkspaceService { async enqueueRunsForWorkspace({ workspaceId, - priorityWorkflowRunId, isCacheMode, }: { workspaceId: string; - priorityWorkflowRunId?: string; isCacheMode: boolean; }) { + const lockAcquired = + await this.workflowThrottlingWorkspaceService.acquireWorkflowEnqueueLock( + workspaceId, + ); + + if (!lockAcquired) { + return; + } + try { const workflowRunRepository = await this.twentyORMGlobalManager.getRepositoryForWorkspace( @@ -66,11 +72,9 @@ export class WorkflowRunEnqueueWorkspaceService { ); if (notStartedRunsCount <= 0) { - if (!isCacheMode) { - await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount( - workspaceId, - ); - } + await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount( + workspaceId, + ); return; } @@ -82,23 +86,6 @@ export class WorkflowRunEnqueueWorkspaceService { 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: { @@ -173,6 +160,10 @@ export class WorkflowRunEnqueueWorkspaceService { `Failed to enqueue workflow runs for workspace: ${workspaceId}`, error, ); + } finally { + await this.workflowThrottlingWorkspaceService.releaseWorkflowEnqueueLock( + workspaceId, + ); } } } 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 index 90323cc70c..2baaaad2e3 100644 --- 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 @@ -12,7 +12,6 @@ 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 { @@ -26,7 +25,7 @@ export class WorkflowThrottlingWorkspaceService { async getRemainingRunsToEnqueueCount(workspaceId: string) { return this.throttlerService.getAvailableTokensCount( - `${workspaceId}-workflow-execution-soft-throttle`, + this.getWorkflowExecutionSoftThrottleCacheKey(workspaceId), this.twentyConfigService.get('WORKFLOW_EXEC_SOFT_THROTTLE_LIMIT'), this.twentyConfigService.get('WORKFLOW_EXEC_SOFT_THROTTLE_TTL'), ); @@ -37,7 +36,7 @@ export class WorkflowThrottlingWorkspaceService { runsToConsume: number, ) { await this.throttlerService.consumeTokens( - `${workspaceId}-workflow-execution-soft-throttle`, + this.getWorkflowExecutionSoftThrottleCacheKey(workspaceId), runsToConsume, this.twentyConfigService.get('WORKFLOW_EXEC_SOFT_THROTTLE_LIMIT'), this.twentyConfigService.get('WORKFLOW_EXEC_SOFT_THROTTLE_TTL'), @@ -46,7 +45,7 @@ export class WorkflowThrottlingWorkspaceService { async throttleOrThrowIfHardLimitReached(workspaceId: string) { await this.throttlerService.tokenBucketThrottleOrThrow( - `${workspaceId}-workflow-execution-hard-throttle`, + this.getWorkflowExecutionHardThrottleCacheKey(workspaceId), 1, this.twentyConfigService.get('WORKFLOW_EXEC_HARD_THROTTLE_LIMIT'), this.twentyConfigService.get('WORKFLOW_EXEC_HARD_THROTTLE_TTL'), @@ -57,12 +56,9 @@ export class WorkflowThrottlingWorkspaceService { workspaceId: string, newlyEnqueuedCount = 1, ): Promise { - const currentCount = - await this.getCurrentWorkflowRunNotStartedCount(workspaceId); - - await this.cacheStorage.set( - getWorkflowRunNotStartedCountCacheKey(workspaceId), - currentCount + newlyEnqueuedCount, + await this.cacheStorage.incrBy( + this.getWorkflowRunNotStartedCountCacheKey(workspaceId), + newlyEnqueuedCount, ); } @@ -70,12 +66,9 @@ export class WorkflowThrottlingWorkspaceService { workspaceId: string, removedFromQueueCount = 1, ): Promise { - const currentCount = - await this.getCurrentWorkflowRunNotStartedCount(workspaceId); - - await this.cacheStorage.set( - getWorkflowRunNotStartedCountCacheKey(workspaceId), - currentCount - removedFromQueueCount, + await this.cacheStorage.incrBy( + this.getWorkflowRunNotStartedCountCacheKey(workspaceId), + -removedFromQueueCount, ); } @@ -123,12 +116,27 @@ export class WorkflowThrottlingWorkspaceService { }); } + async acquireWorkflowEnqueueLock( + workspaceId: string, + ttlMs = 60_000, + ): Promise { + const key = this.getWorkflowEnqueueRunningCacheKey(workspaceId); + + return this.cacheStorage.acquireLock(key, ttlMs); + } + + async releaseWorkflowEnqueueLock(workspaceId: string): Promise { + const key = this.getWorkflowEnqueueRunningCacheKey(workspaceId); + + await this.cacheStorage.releaseLock(key); + } + private async setWorkflowRunNotStartedCount( workspaceId: string, count: number, ): Promise { await this.cacheStorage.set( - getWorkflowRunNotStartedCountCacheKey(workspaceId), + this.getWorkflowRunNotStartedCountCacheKey(workspaceId), count, ); } @@ -136,10 +144,30 @@ export class WorkflowThrottlingWorkspaceService { private async getCurrentWorkflowRunNotStartedCount( workspaceId: string, ): Promise { - const key = getWorkflowRunNotStartedCountCacheKey(workspaceId); + const key = this.getWorkflowRunNotStartedCountCacheKey(workspaceId); const currentCount = (await this.cacheStorage.get(key)) ?? 0; return Math.max(0, currentCount); } + + private getWorkflowRunNotStartedCountCacheKey(workspaceId: string): string { + return `workflow-run-not-started-count:${workspaceId}`; + } + + private getWorkflowEnqueueRunningCacheKey(workspaceId: string): string { + return `workflow-enqueue-running:${workspaceId}`; + } + + private getWorkflowExecutionSoftThrottleCacheKey( + workspaceId: string, + ): string { + return `workflow:execution-soft-throttle:${workspaceId}`; + } + + private getWorkflowExecutionHardThrottleCacheKey( + workspaceId: string, + ): string { + return `workflow:execution-hard-throttle:${workspaceId}`; + } } 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 d59e2cd4bc..cd537549bb 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 @@ -90,13 +90,22 @@ export class WorkflowRunnerWorkspaceService { }); } - return this.createNotStartedWorkflowRunAndTriggerEnqueue({ + if (isManualTrigger) { + return this.enqueueWorkflowRun({ + workspaceId, + workflowVersionId, + initialWorkflowRunId, + source, + payload, + }); + } + + return this.createNotStartedWorkflowRunAndTriggerEnqueueJob({ workspaceId, workflowVersionId, initialWorkflowRunId, source, payload, - isManualTrigger, }); } @@ -272,20 +281,51 @@ export class WorkflowRunnerWorkspaceService { return { workflowRunId }; } - private async createNotStartedWorkflowRunAndTriggerEnqueue({ + private async enqueueWorkflowRun({ + workspaceId, + workflowVersionId, + initialWorkflowRunId, + source, + payload, + }: { + workspaceId: string; + workflowVersionId: string; + initialWorkflowRunId?: string; + source: ActorMetadata; + payload: object; + }) { + const workflowRunId = + await this.workflowRunWorkspaceService.createWorkflowRun({ + workflowVersionId, + workflowRunId: initialWorkflowRunId, + createdBy: source, + status: WorkflowRunStatus.ENQUEUED, + triggerPayload: payload, + }); + + await this.messageQueueService.add( + RunWorkflowJob.name, + { + workspaceId, + workflowRunId, + }, + ); + + return { workflowRunId }; + } + + private async createNotStartedWorkflowRunAndTriggerEnqueueJob({ workspaceId, workflowVersionId, initialWorkflowRunId, source, payload, - isManualTrigger, }: { workspaceId: string; workflowVersionId: string; initialWorkflowRunId?: string; source: ActorMetadata; payload: object; - isManualTrigger: boolean; }) { const workflowRunId = await this.workflowRunWorkspaceService.createWorkflowRun({ @@ -304,7 +344,6 @@ export class WorkflowRunnerWorkspaceService { WorkflowRunEnqueueJob.name, { workspaceId, - priorityWorkflowRunId: isManualTrigger ? workflowRunId : undefined, isCacheMode: true, }, );