From e048d03872cbd2ecc1d5f3f93478f36fca3bdca4 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Tue, 7 Apr 2026 16:09:11 +0200 Subject: [PATCH] Improve workflow crons efficiency (#19381) **Optimize workflow cron jobs: partition workspaces and use raw queries** - Split all 3 workflow cron jobs (WorkflowRunEnqueueCronJob, WorkflowHandleStaledRunsCronJob, WorkflowCleanWorkflowRunsCronJob) to process only 1/10th of workspaces per invocation using minute-based partitioning, reducing per-run load - Replace ORM repository + workspace context loading with raw SQL queries in WorkflowRunEnqueueCronJob and WorkflowHandleStaledRunsCronJob, avoiding costly cache/metadata hydration for a simple existence check --- .../twenty-config/config-variables.ts | 6 +- .../workflow-clean-workflow-runs.cron.job.ts | 40 ++++++++-- .../workflow-handle-staled-runs.cron.job.ts | 80 ++++++++++++------- .../jobs/workflow-run-enqueue.cron.job.ts | 78 +++++++++++------- 4 files changed, 133 insertions(+), 71 deletions(-) diff --git a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts index 537de397700..346517abbb7 100644 --- a/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts +++ b/packages/twenty-server/src/engine/core-modules/twenty-config/config-variables.ts @@ -21,9 +21,6 @@ import { CaptchaDriverType } from 'src/engine/core-modules/captcha/interfaces'; import { CodeInterpreterDriverType } from 'src/engine/core-modules/code-interpreter/code-interpreter.interface'; import { WebSearchDriverType } from 'src/engine/core-modules/web-search/web-search.interface'; import { EmailDriver } from 'src/engine/core-modules/email/enums/email-driver.enum'; -import { type AiModelPreferences } from 'src/engine/metadata-modules/ai/ai-models/types/ai-model-preferences.type'; -import { type AiProvidersConfig } from 'src/engine/metadata-modules/ai/ai-models/types/ai-providers-config.type'; -import { loadDefaultModelPreferences } from 'src/engine/metadata-modules/ai/ai-models/utils/load-default-model-preferences.util'; import { ExceptionHandlerDriver } from 'src/engine/core-modules/exception-handler/interfaces'; import { StorageDriverType } from 'src/engine/core-modules/file-storage/interfaces'; import { LoggerDriverType } from 'src/engine/core-modules/logger/interfaces'; @@ -45,6 +42,9 @@ import { ConfigVariableException, ConfigVariableExceptionCode, } from 'src/engine/core-modules/twenty-config/twenty-config.exception'; +import { type AiModelPreferences } from 'src/engine/metadata-modules/ai/ai-models/types/ai-model-preferences.type'; +import { type AiProvidersConfig } from 'src/engine/metadata-modules/ai/ai-models/types/ai-providers-config.type'; +import { loadDefaultModelPreferences } from 'src/engine/metadata-modules/ai/ai-models/utils/load-default-model-preferences.util'; export class ConfigVariables { @ConfigVariablesMetadata({ diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts index d4e893268f2..70da1ccbb98 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts @@ -4,6 +4,9 @@ import { InjectRepository } from '@nestjs/typeorm'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; import { In, Repository } 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 { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; @@ -25,9 +28,11 @@ import { } from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job'; import { getRunsToCleanFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-runs-to-clean-find-options.util'; -export const CLEAN_WORKFLOW_RUN_CRON_PATTERN = '0 0 * * *'; +export const CLEAN_WORKFLOW_RUN_CRON_PATTERN = '0 */3 * * *'; -const WORKSPACE_BATCH_SIZE = 50; +const LAST_PARTITION_CACHE_KEY = 'workflow-clean-workflow-runs:last-partition'; +const NUMBER_OF_PARTITIONS = 10; +const WORKSPACE_BATCH_SIZE = 10; @Processor(MessageQueue.cronQueue) export class WorkflowCleanWorkflowRunsCronJob { @@ -40,6 +45,8 @@ export class WorkflowCleanWorkflowRunsCronJob { private readonly messageQueueService: MessageQueueService, private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, private readonly exceptionHandlerService: ExceptionHandlerService, + @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) + private readonly cacheStorageService: CacheStorageService, ) {} @Process(WorkflowCleanWorkflowRunsCronJob.name) @@ -50,21 +57,27 @@ export class WorkflowCleanWorkflowRunsCronJob { async handle() { this.logger.log('Starting WorkflowCleanWorkflowRunsCronJob cron'); - const activeWorkspaces = await this.workspaceRepository.find({ + const allActiveWorkspaces = await this.workspaceRepository.find({ where: { activationStatus: WorkspaceActivationStatus.ACTIVE, }, select: ['id'], + order: { id: 'ASC' }, }); + const partition = await this.getAndIncrementPartition(); + const workspacesForThisRun = allActiveWorkspaces.filter( + (_, index) => index % NUMBER_OF_PARTITIONS === partition, + ); + let enqueuedCount = 0; for ( let workspaceIndex = 0; - workspaceIndex < activeWorkspaces.length; + workspaceIndex < workspacesForThisRun.length; workspaceIndex += WORKSPACE_BATCH_SIZE ) { - const batch = activeWorkspaces.slice( + const batch = workspacesForThisRun.slice( workspaceIndex, workspaceIndex + WORKSPACE_BATCH_SIZE, ); @@ -87,7 +100,7 @@ export class WorkflowCleanWorkflowRunsCronJob { } this.logger.log( - `Completed WorkflowCleanWorkflowRunsCronJob cron, enqueued ${enqueuedCount} jobs`, + `Completed WorkflowCleanWorkflowRunsCronJob cron (partition ${partition}/${NUMBER_OF_PARTITIONS}), enqueued ${enqueuedCount} jobs`, ); } @@ -106,6 +119,21 @@ export class WorkflowCleanWorkflowRunsCronJob { return false; } + private async getAndIncrementPartition(): Promise { + const lastPartition = await this.cacheStorageService.get( + LAST_PARTITION_CACHE_KEY, + ); + + const partition = + lastPartition !== undefined + ? (lastPartition + 1) % NUMBER_OF_PARTITIONS + : 0; + + await this.cacheStorageService.set(LAST_PARTITION_CACHE_KEY, partition); + + return partition; + } + private async hasRunsToClean(workspaceId: string): Promise { const authContext = buildSystemAuthContext(workspaceId); diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts index 91d26a65a03..923fd6d3ecb 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts @@ -1,9 +1,12 @@ import { Logger } from '@nestjs/common'; -import { InjectRepository } from '@nestjs/typeorm'; +import { InjectDataSource, InjectRepository } from '@nestjs/typeorm'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; -import { Repository } from 'typeorm'; +import { DataSource, Repository } 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 { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; @@ -12,17 +15,17 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc 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 { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; -import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; -import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; -import { WorkflowRunWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; +import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { STALED_RUNS_THRESHOLD_MS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/staled-runs-threshold'; import { WorkflowHandleStaledRunsJob, WorkflowHandleStaledRunsJobData, } from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job'; -import { getStaledRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.util'; - -export const WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN = '0 * * * *'; +export const WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN = '*/10 * * * *'; +const LAST_PARTITION_CACHE_KEY = 'workflow-handle-staled-runs:last-partition'; +const NUMBER_OF_PARTITIONS = 10; const WORKSPACE_BATCH_SIZE = 50; @Processor(MessageQueue.cronQueue) @@ -30,12 +33,15 @@ export class WorkflowHandleStaledRunsCronJob { private readonly logger = new Logger(WorkflowHandleStaledRunsCronJob.name); constructor( + @InjectDataSource() + private readonly coreDataSource: DataSource, @InjectRepository(WorkspaceEntity) private readonly workspaceRepository: Repository, @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, - private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, private readonly exceptionHandlerService: ExceptionHandlerService, + @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) + private readonly cacheStorageService: CacheStorageService, ) {} @Process(WorkflowHandleStaledRunsCronJob.name) @@ -46,21 +52,27 @@ export class WorkflowHandleStaledRunsCronJob { async handle() { this.logger.log('Starting WorkflowHandleStaledRunsCronJob cron'); - const activeWorkspaces = await this.workspaceRepository.find({ + const allActiveWorkspaces = await this.workspaceRepository.find({ where: { activationStatus: WorkspaceActivationStatus.ACTIVE, }, select: ['id'], + order: { id: 'ASC' }, }); + const partition = await this.getAndIncrementPartition(); + const workspacesForThisRun = allActiveWorkspaces.filter( + (_, index) => index % NUMBER_OF_PARTITIONS === partition, + ); + let enqueuedCount = 0; for ( let workspaceIndex = 0; - workspaceIndex < activeWorkspaces.length; + workspaceIndex < workspacesForThisRun.length; workspaceIndex += WORKSPACE_BATCH_SIZE ) { - const batch = activeWorkspaces.slice( + const batch = workspacesForThisRun.slice( workspaceIndex, workspaceIndex + WORKSPACE_BATCH_SIZE, ); @@ -83,7 +95,7 @@ export class WorkflowHandleStaledRunsCronJob { } this.logger.log( - `Completed WorkflowHandleStaledRunsCronJob cron, enqueued ${enqueuedCount} jobs`, + `Completed WorkflowHandleStaledRunsCronJob cron (partition ${partition}/${NUMBER_OF_PARTITIONS}), enqueued ${enqueuedCount} jobs`, ); } @@ -102,24 +114,30 @@ export class WorkflowHandleStaledRunsCronJob { return false; } - private async hasStaledRuns(workspaceId: string): Promise { - const authContext = buildSystemAuthContext(workspaceId); - - return this.globalWorkspaceOrmManager.executeInWorkspaceContext( - async () => { - const workflowRunRepository = - await this.globalWorkspaceOrmManager.getRepository( - workspaceId, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); - - return workflowRunRepository.exists({ - where: getStaledRunsFindOptions(), - }); - }, - authContext, - { lite: true }, + private async getAndIncrementPartition(): Promise { + const lastPartition = await this.cacheStorageService.get( + LAST_PARTITION_CACHE_KEY, ); + + const partition = + lastPartition !== undefined + ? (lastPartition + 1) % NUMBER_OF_PARTITIONS + : 0; + + await this.cacheStorageService.set(LAST_PARTITION_CACHE_KEY, partition); + + return partition; + } + + private async hasStaledRuns(workspaceId: string): Promise { + const schemaName = getWorkspaceSchemaName(workspaceId); + const thresholdDate = new Date(Date.now() - STALED_RUNS_THRESHOLD_MS); + + const result = await this.coreDataSource.query( + `SELECT 1 FROM ${schemaName}."workflowRun" WHERE "status" = $1 AND ("enqueuedAt" < $2 OR "enqueuedAt" IS NULL) LIMIT 1`, + [WorkflowRunStatus.ENQUEUED, thresholdDate], + ); + + return result.length > 0; } } 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 1f8bf00ae13..d4a22c1ee79 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 @@ -1,9 +1,12 @@ import { Logger } from '@nestjs/common'; -import { InjectRepository } from '@nestjs/typeorm'; +import { InjectDataSource, InjectRepository } from '@nestjs/typeorm'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; -import { Repository } from 'typeorm'; +import { DataSource, Repository } 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 { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; @@ -12,17 +15,16 @@ import { Processor } from 'src/engine/core-modules/message-queue/decorators/proc 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 { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity'; -import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; -import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util'; -import { WorkflowRunWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; -import { NOT_STARTED_RUNS_FIND_OPTIONS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options'; +import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; +import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { WorkflowRunEnqueueJob, WorkflowRunEnqueueJobData, } from 'src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-run-enqueue.job'; +export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '* * * * *'; -export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '*/5 * * * *'; - +const LAST_PARTITION_CACHE_KEY = 'workflow-run-enqueue:last-partition'; +const NUMBER_OF_PARTITIONS = 10; const WORKSPACE_BATCH_SIZE = 10; @Processor(MessageQueue.cronQueue) @@ -30,12 +32,15 @@ export class WorkflowRunEnqueueCronJob { private readonly logger = new Logger(WorkflowRunEnqueueCronJob.name); constructor( + @InjectDataSource() + private readonly coreDataSource: DataSource, @InjectRepository(WorkspaceEntity) private readonly workspaceRepository: Repository, @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, - private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, private readonly exceptionHandlerService: ExceptionHandlerService, + @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) + private readonly cacheStorageService: CacheStorageService, ) {} @Process(WorkflowRunEnqueueCronJob.name) @@ -46,21 +51,27 @@ export class WorkflowRunEnqueueCronJob { async handle() { this.logger.log('Starting WorkflowRunEnqueueCronJob cron'); - const activeWorkspaces = await this.workspaceRepository.find({ + const allActiveWorkspaces = await this.workspaceRepository.find({ where: { activationStatus: WorkspaceActivationStatus.ACTIVE, }, select: ['id'], + order: { id: 'ASC' }, }); + const partition = await this.getAndIncrementPartition(); + const workspacesForThisRun = allActiveWorkspaces.filter( + (_, index) => index % NUMBER_OF_PARTITIONS === partition, + ); + let enqueuedCount = 0; for ( let workspaceIndex = 0; - workspaceIndex < activeWorkspaces.length; + workspaceIndex < workspacesForThisRun.length; workspaceIndex += WORKSPACE_BATCH_SIZE ) { - const batch = activeWorkspaces.slice( + const batch = workspacesForThisRun.slice( workspaceIndex, workspaceIndex + WORKSPACE_BATCH_SIZE, ); @@ -83,7 +94,7 @@ export class WorkflowRunEnqueueCronJob { } this.logger.log( - `Completed WorkflowRunEnqueueCronJob cron, enqueued ${enqueuedCount} jobs`, + `Completed WorkflowRunEnqueueCronJob cron (partition ${partition}/${NUMBER_OF_PARTITIONS}), enqueued ${enqueuedCount} jobs`, ); } @@ -102,24 +113,29 @@ export class WorkflowRunEnqueueCronJob { return false; } - private async hasNotStartedRuns(workspaceId: string): Promise { - const authContext = buildSystemAuthContext(workspaceId); - - return this.globalWorkspaceOrmManager.executeInWorkspaceContext( - async () => { - const workflowRunRepository = - await this.globalWorkspaceOrmManager.getRepository( - workspaceId, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); - - return workflowRunRepository.exists({ - where: NOT_STARTED_RUNS_FIND_OPTIONS, - }); - }, - authContext, - { lite: true }, + private async getAndIncrementPartition(): Promise { + const lastPartition = await this.cacheStorageService.get( + LAST_PARTITION_CACHE_KEY, ); + + const partition = + lastPartition !== undefined + ? (lastPartition + 1) % NUMBER_OF_PARTITIONS + : 0; + + await this.cacheStorageService.set(LAST_PARTITION_CACHE_KEY, partition); + + return partition; + } + + private async hasNotStartedRuns(workspaceId: string): Promise { + const schemaName = getWorkspaceSchemaName(workspaceId); + + const result = await this.coreDataSource.query( + `SELECT 1 FROM ${schemaName}."workflowRun" WHERE "status" = $1 LIMIT 1`, + [WorkflowRunStatus.NOT_STARTED], + ); + + return result.length > 0; } }