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

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<number> {
+    if (this.isRedisCache()) {
+      return (this.cache as RedisCache).store.client.incrBy(
+        this.getKey(key),
+        increment,
+      );
+    }
+
+    const current = (await this.get<number>(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';

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<WorkspaceEntity>,
-    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,
     });

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<void> {
     await this.WorkflowRunEnqueueWorkspaceService.enqueueRunsForWorkspace({
       workspaceId,
-      priorityWorkflowRunId,
       isCacheMode,
     });
   }

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util.ts

@@ -1,3 +0,0 @@
-export const getWorkflowRunNotStartedCountCacheKey = (
-  workspaceId: string,
-): string => `workflow-run-not-started-count:${workspaceId}`;

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,
+      );
     }
   }
 }

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<void> {
-    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<void> {
-    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<boolean> {
+    const key = this.getWorkflowEnqueueRunningCacheKey(workspaceId);
+
+    return this.cacheStorage.acquireLock(key, ttlMs);
+  }
+
+  async releaseWorkflowEnqueueLock(workspaceId: string): Promise<void> {
+    const key = this.getWorkflowEnqueueRunningCacheKey(workspaceId);
+
+    await this.cacheStorage.releaseLock(key);
+  }
+
   private async setWorkflowRunNotStartedCount(
     workspaceId: string,
     count: number,
   ): Promise<void> {
     await this.cacheStorage.set(
-      getWorkflowRunNotStartedCountCacheKey(workspaceId),
+      this.getWorkflowRunNotStartedCountCacheKey(workspaceId),
       count,
     );
   }
@@ -136,10 +144,30 @@ export class WorkflowThrottlingWorkspaceService {
   private async getCurrentWorkflowRunNotStartedCount(
     workspaceId: string,
   ): Promise<number> {
-    const key = getWorkflowRunNotStartedCountCacheKey(workspaceId);
+    const key = this.getWorkflowRunNotStartedCountCacheKey(workspaceId);
 
     const currentCount = (await this.cacheStorage.get<number>(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}`;
+  }
 }

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<RunWorkflowJobData>(
+      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,
       },
     );