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

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<number> {
+    const lastPartition = await this.cacheStorageService.get<number>(
+      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<boolean> {
     const authContext = buildSystemAuthContext(workspaceId);
 

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<WorkspaceEntity>,
     @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<boolean> {
-    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<number> {
+    const lastPartition = await this.cacheStorageService.get<number>(
+      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<boolean> {
+    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;
   }
 }

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<WorkspaceEntity>,
     @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<boolean> {
-    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<number> {
+    const lastPartition = await this.cacheStorageService.get<number>(
+      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<boolean> {
+    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;
   }
 }