Improve workflow throttling logic by thomtrp · Pull Request #16260 · twentyhq/twenty

thomtrp · 2025-12-02T15:01:44Z

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

...-runner/workflow-run-queue/workspace-services/workflow-not-started-runs.workspace-service.ts

greptile-apps · 2025-12-02T15:12:37Z

Greptile Overview

Greptile Summary

This PR implements a two-tier throttling system for workflow execution with hard and soft limits:

Key Changes:

Hard throttle (5000 workflows/hour): New workflows are marked as FAILED with error message when exceeded
Soft throttle (100 workflows/min): New workflows are marked as NOT_STARTED and queued for later execution when exceeded
Manual triggers bypass soft throttle but respect hard throttle
New WorkflowNotStartedRunsWorkspaceService manages NOT_STARTED workflows with cache-based counting
Enhanced ThrottlerService now returns remaining token count
WorkflowEnqueueAwaitingRunsJob processes queued workflows when capacity becomes available
Added metrics tracking for throttled workflows (WorkflowRunSoftThrottled, WorkflowRunHardThrottled)

Issues Found:

Potential race condition where remainingRunsToEnqueueCount becomes stale between throttle check and awaiting runs enqueue decision
Manual trigger bypass logic may prevent enqueueing awaiting runs even when capacity is available

Confidence Score: 3/5

This PR has logical issues that could affect workflow execution reliability under high load
The implementation introduces good architectural patterns (two-tier throttling, metrics, cache-based counting) but contains race conditions with stale token counts and logic issues with manual trigger bypass that could prevent proper workflow queue processing
packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts requires attention for race condition and manual trigger logic fixes

Important Files Changed

File Analysis

Filename	Score	Overview
packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts	3/5	Refactored throttling logic with separate hard/soft limits, added metrics tracking, and new `NOT_STARTED` workflow run status. Contains potential race condition with stale `remainingRunsToEnqueueCount` and manual trigger bypass logic issue.
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-not-started-runs.workspace-service.ts	4/5	New service managing `NOT_STARTED` workflow runs with cache-based counting and throttling integration. Provides methods for tracking, recomputing, and throttling workflow execution with hard/soft limits.
packages/twenty-server/src/engine/core-modules/throttler/throttler.service.ts	4/5	Enhanced token bucket throttler to return remaining tokens and added `getAvailableTokensCount` method. Clean refactoring with minor timing precision concern.
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-enqueue-awaiting-runs.workspace-service.ts	4/5	New service for processing `NOT_STARTED` workflows and enqueueing them when capacity is available. Includes proper error handling and metrics tracking.

Sequence Diagram

sequenceDiagram
    participant Client
    participant WorkflowRunner as WorkflowRunnerService
    participant NotStartedSvc as NotStartedRunsService
    participant Throttler as ThrottlerService
    participant Cache as Redis Cache
    participant DB as Database
    participant Queue as Message Queue

    Client->>WorkflowRunner: run(workflowVersionId, payload)
    WorkflowRunner->>WorkflowRunner: Check if manual trigger
    
    WorkflowRunner->>NotStartedSvc: throttleOrThrowIfHardLimitReached()
    NotStartedSvc->>Throttler: tokenBucketThrottleOrThrow(hard-throttle key, 1, 5000, 3600000)
    Throttler->>Cache: get(hard-throttle tokens)
    Cache-->>Throttler: current token state
    alt Hard limit exceeded (>5000/hour)
        Throttler-->>NotStartedSvc: throw ThrottlerException
        NotStartedSvc-->>WorkflowRunner: throw
        WorkflowRunner->>WorkflowRunner: createFailedWorkflowRun()
        WorkflowRunner->>DB: create workflow run (status=FAILED, error="Throttle limit reached")
        WorkflowRunner-->>Client: return {workflowRunId}
    else Hard limit OK
        Throttler->>Cache: set(hard-throttle tokens - 1)
        Throttler-->>NotStartedSvc: return remaining tokens
    end

    WorkflowRunner->>NotStartedSvc: throttleAndReturnRemainingRunsToEnqueueCount()
    NotStartedSvc->>Throttler: tokenBucketThrottleOrThrow(soft-throttle key, 1, 100, 60000)
    Throttler->>Cache: get(soft-throttle tokens)
    Cache-->>Throttler: current token state
    alt Soft limit exceeded (>100/min) AND NOT manual trigger
        Throttler-->>NotStartedSvc: throw ThrottlerException
        NotStartedSvc-->>WorkflowRunner: throw
        WorkflowRunner->>WorkflowRunner: createNotStartedWorkflowRun()
        WorkflowRunner->>DB: create workflow run (status=NOT_STARTED)
        WorkflowRunner->>NotStartedSvc: increaseWorkflowRunNotStartedCount()
        NotStartedSvc->>Cache: increment NOT_STARTED count
        WorkflowRunner-->>Client: return {workflowRunId}
    else Soft limit OK OR manual trigger
        Throttler->>Cache: set(soft-throttle tokens - 1)
        Throttler-->>NotStartedSvc: return remaining tokens
        NotStartedSvc-->>WorkflowRunner: return remainingCount
    end

    WorkflowRunner->>WorkflowRunner: enqueueWorkflowRunAndPotentialAwaitingRuns()
    WorkflowRunner->>DB: create workflow run (status=ENQUEUED)
    WorkflowRunner->>Queue: add(RunWorkflowJob, {workflowRunId})
    
    WorkflowRunner->>NotStartedSvc: getNotStartedRunsCountFromCache()
    NotStartedSvc->>Cache: get(NOT_STARTED count)
    Cache-->>NotStartedSvc: current count
    NotStartedSvc-->>WorkflowRunner: return count
    
    alt remainingCount > 0 AND notStartedCount > 0
        WorkflowRunner->>Queue: add(WorkflowEnqueueAwaitingRunsJob, {workspaceId})
        Note over Queue: Background job will process<br/>NOT_STARTED workflows later
    end
    
    WorkflowRunner-->>Client: return {workflowRunId}

greptile-apps

_{25 files reviewed, 1 comment}

_{Edit Code Review Agent Settings | Greptile}

...dules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-enqueue-awaiting-runs.job.ts

github-actions · 2025-12-02T15:14:18Z

🚀 Preview Environment Ready!

Your preview environment is available at: http://bore.pub:55106

This environment will automatically shut down when the PR is closed or after 5 hours.

thomtrp · 2025-12-02T15:15:04Z

@greptile-ai

greptile-apps

_{25 files reviewed, 3 comments}

_{Edit Code Review Agent Settings | Greptile}

...src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts

packages/twenty-server/src/engine/core-modules/throttler/throttler.service.ts

charlesBochet · 2025-12-02T15:47:45Z

...dules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-enqueue-awaiting-runs.job.ts

+
+  @Process(WorkflowEnqueueAwaitingRunsJob.name)
+  async handle({ workspaceId }: { workspaceId: string }): Promise<void> {
+    await this.workflowEnqueueAwaitingRunsWorkspaceService.enqueueRuns({


charlesBochet · 2025-12-02T15:48:18Z

...orkflow-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}`;


why not using workspace cache service?

charlesBochet · 2025-12-02T15:50:40Z

...er/workflow-run-queue/workspace-services/workflow-enqueue-awaiting-runs.workspace-service.ts

-        const remainingWorkflowRunToEnqueueCount =
-          await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromDatabase(
+        const notStartedRunsCount =
+          await this.workflowNotStartedRunsWorkspaceService.getNotStartedRunsCountFromDatabase(


we should maintain in cache

charlesBochet · 2025-12-02T15:51:34Z

...unner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts

@@ -50,7 +50,7 @@ export class WorkflowHandleStaledRunsWorkspaceService {
          },
        );

-        await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
+        await this.workflowNotStartedRunsWorkspaceService.recomputeWorkflowRunNotStartedCount(


not useful anymore?

charlesBochet · 2025-12-02T16:12:03Z

...dules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-enqueue-awaiting-runs.job.ts

+import { WorkflowEnqueueAwaitingRunsWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-enqueue-awaiting-runs.workspace-service';
+
+@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST })
+export class WorkflowEnqueueAwaitingRunsJob {


workflow enqueue not started Job
(i would also add an optional workflowRunId? )

charlesBochet · 2025-12-02T16:12:47Z

...dules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-enqueue-awaiting-runs.job.ts

+@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST })
+export class WorkflowEnqueueAwaitingRunsJob {
+  constructor(
+    private readonly workflowEnqueueAwaitingRunsWorkspaceService: WorkflowEnqueueAwaitingRunsWorkspaceService,


workflowEnqueueRunWorkspaceService

charlesBochet · 2025-12-02T16:13:09Z

...-runner/workflow-run-queue/workspace-services/workflow-not-started-runs.workspace-service.ts

+import { getWorkflowRunNotStartedCountCacheKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-workflow-run-not-started-count-cache-key.util';
+
+@Injectable()
+export class WorkflowNotStartedRunsWorkspaceService {


workflow throttling

charlesBochet · 2025-12-02T16:13:56Z

...src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts

+        triggerPayload: payload,
+      });
+
+    await this.enqueueWorkflowRun(workspaceId, workflowRunId);


remove this

charlesBochet · 2025-12-02T16:14:10Z

...src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts

+
+    if (remainingRunsToEnqueueCount > 0 && currentNotStartedRunsCount > 0) {
+      await this.messageQueueService.add<{ workspaceId: string }>(
+        WorkflowEnqueueAwaitingRunsJob.name,


pass prioritiarty workflowRunId if needed here

charlesBochet · 2025-12-02T16:14:53Z

...src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts

@@ -244,4 +253,136 @@ export class WorkflowRunnerWorkspaceService {
      status: newStatus,
    };
  }
+
+  private async checkThrottleLimits(


check only hard limit

no need to handle soft here

…rkflow-run-queue/jobs/workflow-enqueue-awaiting-runs.job.ts Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

charlesBochet · 2025-12-03T07:38:53Z

...odules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts

        workflowRunId,
      });

      await this.messageQueueService.add<RunWorkflowJobData>(


I feel this should also go through workflow-run-enqueue job

The enqueue notion means NOT_STARTED => ENQUEUE. The service only looks for not started workflows. In case of a running workflow, like here, we still need to run the job separately. So three cases: delays (here), forms and when a workflow has two many steps, we cut and re-send the job in the queue

charlesBochet · 2025-12-03T07:39:17Z

...ver/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job.ts

+
+export type WorkflowRunEnqueueJobData = {
+  workspaceId: string;
+  workflowRunId?: string;


prioritaryWorkflowRunId

twenty-eng-sync · 2025-12-03T09:01:53Z

Hey @thomtrp! After you've done the QA of your Pull Request, you can mark it as done here. Thank you!

sentry bot reviewed Dec 2, 2025

View reviewed changes

...-runner/workflow-run-queue/workspace-services/workflow-not-started-runs.workspace-service.ts Outdated Show resolved Hide resolved

thomtrp force-pushed the tt-workflow-throttling-v2 branch 2 times, most recently from 800feaa to 52db40e Compare December 2, 2025 15:07

greptile-apps bot reviewed Dec 2, 2025

View reviewed changes

...dules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-enqueue-awaiting-runs.job.ts Outdated Show resolved Hide resolved

greptile-apps bot reviewed Dec 2, 2025

View reviewed changes

charlesBochet reviewed Dec 2, 2025

View reviewed changes

charlesBochet assigned thomtrp Dec 2, 2025

charlesBochet added the -PR: awaiting author label Dec 2, 2025

thomtrp and others added 4 commits December 2, 2025 17:48

Improve throttling logic

fc93512

Update packages/twenty-server/src/modules/workflow/workflow-runner/wo…

6b8e8d9

…rkflow-run-queue/jobs/workflow-enqueue-awaiting-runs.job.ts Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

Clean test

3da87a5

Move all queueing logic into only one service

acba2b0

thomtrp force-pushed the tt-workflow-throttling-v2 branch from 32ab36d to acba2b0 Compare December 2, 2025 17:40

charlesBochet approved these changes Dec 3, 2025

View reviewed changes

charlesBochet reviewed Dec 3, 2025

View reviewed changes

Rename priorityWorkflowRunId

c57f66a

thomtrp enabled auto-merge (squash) December 3, 2025 08:59

thomtrp merged commit a0f196e into main Dec 3, 2025
53 checks passed

thomtrp deleted the tt-workflow-throttling-v2 branch December 3, 2025 09:01

Conversation

thomtrp commented Dec 2, 2025

Uh oh!

Uh oh!

greptile-apps bot commented Dec 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Greptile Overview

Greptile Summary

Confidence Score: 3/5

Important Files Changed

Sequence Diagram

Uh oh!

greptile-apps bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

github-actions bot commented Dec 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

thomtrp commented Dec 2, 2025

Uh oh!

greptile-apps bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

twenty-eng-sync bot commented Dec 3, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

greptile-apps bot commented Dec 2, 2025 •

edited

Loading

github-actions bot commented Dec 2, 2025 •

edited

Loading