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>
2025-12-03 10:01:36 +01:00
parent 288db78abd
commit a0f196e871
25 changed files with 538 additions and 386 deletions
@@ -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,
@@ -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',
@@ -20,18 +20,14 @@ export class ThrottlerService {
    tokensToConsume: number,
    maxTokens: number,
    timeWindow: number,
-  ): Promise<void> {
+  ): Promise<number> {
    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<number> {
+    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);
  }
 }
@@ -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,
@@ -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,
@@ -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: [
@@ -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(
@@ -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,
-    );
  }
 }
@@ -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,
@@ -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<void> {
-    const { workspaceIds } = options;
-
-    await this.workflowRunEnqueueWorkspaceService.enqueueRuns({
-      workspaceIds,
-    });
-  }
-}
@@ -1 +0,0 @@
-export const DEFAULT_WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100;
@@ -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<void> {
    await this.messageQueueService.addCron({
-      jobName: WorkflowRunEnqueueJob.name,
+      jobName: WorkflowRunEnqueueCronJob.name,
      data: undefined,
      options: {
        repeat: {
@@ -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<WorkspaceEntity>,
-    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,
    });
  }
 }
@@ -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<void> {
+    await this.WorkflowRunEnqueueWorkspaceService.enqueueRunsForWorkspace({
+      workspaceId,
+      priorityWorkflowRunId,
+      isCacheMode,
+    });
+  }
+}
@@ -1,3 +0,0 @@
-export const getWorkflowRunQueuedCountCacheKey = (
-  workspaceId: string,
-): string => `workflow-run-queued-count:${workspaceId}`;
@@ -1,3 +0,0 @@
-export const getWorkflowRunQueueThrottleLimitKey = (
-  workspaceId: string,
-): string => `workflow-run-queue-throttle-limit:${workspaceId}`;
@@ -0,0 +1,3 @@
+export const getWorkflowRunNotStartedCountCacheKey = (
+  workspaceId: string,
+): string => `workflow-run-not-started-count:${workspaceId}`;
@@ -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,
@@ -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) {
@@ -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<RunWorkflowJobData>(
-            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<RunWorkflowJobData>(
+          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,
+      );
    }
  }
 }
@@ -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<void> {
-    const currentCount =
-      await this.getCurrentWorkflowRunQueuedCount(workspaceId);
-
-    await this.cacheStorage.set(
-      getWorkflowRunQueuedCountCacheKey(workspaceId),
-      currentCount + newlyEnqueuedCount,
-    );
-  }
-
-  async decreaseWorkflowRunQueuedCount(
-    workspaceId: string,
-    removedFromQueueCount = 1,
-  ): Promise<void> {
-    const currentCount =
-      await this.getCurrentWorkflowRunQueuedCount(workspaceId);
-
-    await this.cacheStorage.set(
-      getWorkflowRunQueuedCountCacheKey(workspaceId),
-      currentCount - removedFromQueueCount,
-    );
-  }
-
-  async recomputeWorkflowRunQueuedCount(workspaceId: string): Promise<void> {
-    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<number> {
-    const currentCount =
-      await this.getCurrentWorkflowRunQueuedCount(workspaceId);
-    const throttleLimit =
-      await this.getWorkflowRunQueueThrottleLimit(workspaceId);
-
-    return throttleLimit - currentCount;
-  }
-
-  async getRemainingRunsToEnqueueCountFromDatabase(
-    workspaceId: string,
-  ): Promise<number> {
-    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<void> {
-    await this.cacheStorage.set(
-      getWorkflowRunQueuedCountCacheKey(workspaceId),
-      count,
-    );
-  }
-
-  private async getCurrentWorkflowRunQueuedCount(
-    workspaceId: string,
-  ): Promise<number> {
-    const key = getWorkflowRunQueuedCountCacheKey(workspaceId);
-
-    const currentCount = (await this.cacheStorage.get<number>(key)) ?? 0;
-
-    return Math.max(0, currentCount);
-  }
-
-  private async getWorkflowRunQueueThrottleLimit(
-    workspaceId: string,
-  ): Promise<number> {
-    const key = getWorkflowRunQueueThrottleLimitKey(workspaceId);
-
-    const throttleLimit = (await this.cacheStorage.get<number>(key)) ?? 0;
-
-    return Math.max(DEFAULT_WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT, throttleLimit);
-  }
-}
@@ -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<void> {
+    const currentCount =
+      await this.getCurrentWorkflowRunNotStartedCount(workspaceId);
+
+    await this.cacheStorage.set(
+      getWorkflowRunNotStartedCountCacheKey(workspaceId),
+      currentCount + newlyEnqueuedCount,
+    );
+  }
+
+  async decreaseWorkflowRunNotStartedCount(
+    workspaceId: string,
+    removedFromQueueCount = 1,
+  ): Promise<void> {
+    const currentCount =
+      await this.getCurrentWorkflowRunNotStartedCount(workspaceId);
+
+    await this.cacheStorage.set(
+      getWorkflowRunNotStartedCountCacheKey(workspaceId),
+      currentCount - removedFromQueueCount,
+    );
+  }
+
+  async recomputeWorkflowRunNotStartedCount(
+    workspaceId: string,
+  ): Promise<void> {
+    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<number> {
+    return this.getCurrentWorkflowRunNotStartedCount(workspaceId);
+  }
+
+  async getNotStartedRunsCountFromDatabase(
+    workspaceId: string,
+  ): Promise<number> {
+    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<void> {
+    await this.cacheStorage.set(
+      getWorkflowRunNotStartedCountCacheKey(workspaceId),
+      count,
+    );
+  }
+
+  private async getCurrentWorkflowRunNotStartedCount(
+    workspaceId: string,
+  ): Promise<number> {
+    const key = getWorkflowRunNotStartedCountCacheKey(workspaceId);
+
+    const currentCount = (await this.cacheStorage.get<number>(key)) ?? 0;
+
+    return Math.max(0, currentCount);
+  }
+}
@@ -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,
    };
  }

@@ -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,
@@ -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<RunWorkflowJobData>(
+      RunWorkflowJob.name,
+      {
+        workspaceId,
+        workflowRunId,
+        lastExecutedStepId,
+      },
    );
  }

@@ -179,24 +183,6 @@ export class WorkflowRunnerWorkspaceService {
    });
  }

-  async enqueueWorkflowRun(
-    workspaceId: string,
-    workflowRunId: string,
-    lastExecutedStepId?: string,
-  ) {
-    await this.messageQueueService.add<RunWorkflowJobData>(
-      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<boolean> {
+    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<WorkflowRunEnqueueJobData>(
+      WorkflowRunEnqueueJob.name,
+      {
+        workspaceId,
+        priorityWorkflowRunId: isManualTrigger ? workflowRunId : undefined,
+        isCacheMode: true,
+      },
+    );
+
+    return { workflowRunId };
+  }
 }
				`@@ -1 +0,0 @@`
				`export const DEFAULT_WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100;`