diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/default-workflow-run-queue-throttle-limit.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/default-workflow-run-queue-throttle-limit.ts new file mode 100644 index 0000000000..86c818e318 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/default-workflow-run-queue-throttle-limit.ts @@ -0,0 +1 @@ +export const DEFAULT_WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts deleted file mode 100644 index 9b8ac7e683..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts +++ /dev/null @@ -1 +0,0 @@ -export const WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-queue-throttle-limit-key.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-queue-throttle-limit-key.util.ts new file mode 100644 index 0000000000..1556260922 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-queue-throttle-limit-key.util.ts @@ -0,0 +1,3 @@ +export const getWorkflowRunQueueThrottleLimitKey = ( + workspaceId: string, +): string => `workflow-run-queue-throttle-limit:${workspaceId}`; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts index 758721bade..dd723f8107 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts @@ -10,8 +10,9 @@ import { WorkflowRunStatus, WorkflowRunWorkspaceEntity, } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; -import { WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit'; +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 { @@ -74,8 +75,10 @@ export class WorkflowRunQueueWorkspaceService { ): Promise { const currentCount = await this.getCurrentWorkflowRunQueuedCount(workspaceId); + const throttleLimit = + await this.getWorkflowRunQueueThrottleLimit(workspaceId); - return WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount; + return throttleLimit - currentCount; } async getRemainingRunsToEnqueueCountFromDatabase( @@ -94,7 +97,10 @@ export class WorkflowRunQueueWorkspaceService { }, }); - return WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount; + const throttleLimit = + await this.getWorkflowRunQueueThrottleLimit(workspaceId); + + return throttleLimit - currentCount; } private async setWorkflowRunQueuedCount( @@ -116,4 +122,14 @@ export class WorkflowRunQueueWorkspaceService { return Math.max(0, currentCount); } + + private async getWorkflowRunQueueThrottleLimit( + workspaceId: string, + ): Promise { + const key = getWorkflowRunQueueThrottleLimitKey(workspaceId); + + const throttleLimit = (await this.cacheStorage.get(key)) ?? 0; + + return Math.max(DEFAULT_WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT, throttleLimit); + } }