Improve workflow queue cron reliability (#13818)

Improve workflow enqueue cron : instead of relying on the cache to know
how many workflows we can enqueue, query the DB. Then set the cache and
process the not started workflows.

Also adding a second cron that will look for workflows enqueued one hour
ago or more and put these back in the not started status. This will
allow the first cron to start these again.

packages/twenty-server/src/database/commands/cron-register-all.command.ts

@@ -2,6 +2,7 @@ import { Logger } from '@nestjs/common';
 
 import { Command, CommandRunner } from 'nest-commander';
 
+import { CheckCustomDomainValidRecordsCronCommand } from 'src/engine/core-modules/domain-manager/crons/commands/check-custom-domain-valid-records.cron.command';
 import { CleanupOrphanedFilesCronCommand } from 'src/engine/core-modules/file/crons/commands/cleanup-orphaned-files.cron.command';
 import { CalendarEventListFetchCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-event-list-fetch.cron.command';
 import { CalendarEventsImportCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-import.cron.command';
@@ -9,8 +10,9 @@ import { CalendarOngoingStaleCronCommand } from 'src/modules/calendar/calendar-e
 import { MessagingMessageListFetchCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-message-list-fetch.cron.command';
 import { MessagingMessagesImportCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command';
 import { MessagingOngoingStaleCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-ongoing-stale.cron.command';
-import { CronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command';
-import { CheckCustomDomainValidRecordsCronCommand } from 'src/engine/core-modules/domain-manager/crons/commands/check-custom-domain-valid-records.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 { WorkflowCronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command';
 
 @Command({
   name: 'cron:register:all',
@@ -26,9 +28,11 @@ export class CronRegisterAllCommand extends CommandRunner {
     private readonly calendarEventListFetchCronCommand: CalendarEventListFetchCronCommand,
     private readonly calendarEventsImportCronCommand: CalendarEventsImportCronCommand,
     private readonly calendarOngoingStaleCronCommand: CalendarOngoingStaleCronCommand,
-    private readonly cronTriggerCronCommand: CronTriggerCronCommand,
+    private readonly workflowCronTriggerCronCommand: WorkflowCronTriggerCronCommand,
     private readonly cleanupOrphanedFilesCronCommand: CleanupOrphanedFilesCronCommand,
     private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand,
+    private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand,
+    private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand,
   ) {
     super();
   }
@@ -61,7 +65,6 @@ export class CronRegisterAllCommand extends CommandRunner {
         name: 'CalendarOngoingStale',
         command: this.calendarOngoingStaleCronCommand,
       },
-      { name: 'CronTrigger', command: this.cronTriggerCronCommand },
       {
         name: 'CleanupOrphanedFiles',
         command: this.cleanupOrphanedFilesCronCommand,
@@ -70,6 +73,18 @@ export class CronRegisterAllCommand extends CommandRunner {
         name: 'CheckCustomDomainValidRecords',
         command: this.checkCustomDomainValidRecordsCronCommand,
       },
+      {
+        name: 'WorkflowCronTrigger',
+        command: this.workflowCronTriggerCronCommand,
+      },
+      {
+        name: 'WorkflowRunEnqueue',
+        command: this.workflowRunEnqueueCronCommand,
+      },
+      {
+        name: 'WorkflowHandleStaledRuns',
+        command: this.workflowHandleStaledRunsCronCommand,
+      },
     ];
 
     let successCount = 0;

packages/twenty-server/src/database/commands/database-command.module.ts

@@ -7,6 +7,7 @@ import { UpgradeVersionCommandModule } from 'src/database/commands/upgrade-versi
 import { MigrateViewsToCoreCommand } from 'src/database/commands/views-migration/migrate-views-to-core.command';
 import { TypeORMModule } from 'src/database/typeorm/typeorm.module';
 import { ApiKeyModule } from 'src/engine/core-modules/api-key/api-key.module';
+import { DomainManagerModule } from 'src/engine/core-modules/domain-manager/domain-manager.module';
 import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { FileModule } from 'src/engine/core-modules/file/file.module';
 import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
@@ -18,24 +19,22 @@ import { DevSeederModule } from 'src/engine/workspace-manager/dev-seeder/dev-see
 import { WorkspaceManagerModule } from 'src/engine/workspace-manager/workspace-manager.module';
 import { CalendarEventImportManagerModule } from 'src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module';
 import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module';
+import { WorkflowRunQueueModule } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module';
 import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module';
-import { DomainManagerModule } from 'src/engine/core-modules/domain-manager/domain-manager.module';
 
 import { DataSeedWorkspaceCommand } from './data-seed-dev-workspace.command';
 
 @Module({
   imports: [
     UpgradeVersionCommandModule,
-
     TypeOrmModule.forFeature([Workspace], 'core'),
-
     // Cron command dependencies
     MessagingImportManagerModule,
     CalendarEventImportManagerModule,
     AutomatedTriggerModule,
     FileModule,
     DomainManagerModule,
-
+    WorkflowRunQueueModule,
     // Data seeding dependencies
     TypeORMModule,
     FieldMetadataModule,

packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts

@@ -466,6 +466,7 @@ export const WORKFLOW_RUN_STANDARD_FIELD_IDS = {
   name: '20202020-b840-4253-aef9-4e5013694587',
   workflowVersion: '20202020-2f52-4ba8-8dc4-d0d6adb9578d',
   workflow: '20202020-8c57-4e7f-84f5-f373f68e1b82',
+  enqueuedAt: '20202020-f1e3-4de1-a461-b5c4fdbc861d',
   startedAt: '20202020-a234-4e2d-bd15-85bcea6bb183',
   endedAt: '20202020-e1c1-4b6b-bbbd-b2beaf2e159e',
   status: '20202020-6b3e-4f9c-8c2b-2e5b8e6d6f3b',

packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts

@@ -90,6 +90,16 @@ export class WorkflowRunWorkspaceEntity extends BaseWorkspaceEntity {
   })
   name: string;
 
+  @WorkspaceField({
+    standardId: WORKFLOW_RUN_STANDARD_FIELD_IDS.enqueuedAt,
+    type: FieldMetadataType.DATE_TIME,
+    label: msg`Workflow run enqueued at`,
+    description: msg`Workflow run enqueued at`,
+    icon: 'IconHistory',
+  })
+  @WorkspaceIsNullable()
+  enqueuedAt: Date | null;
+
   @WorkspaceField({
     standardId: WORKFLOW_RUN_STANDARD_FIELD_IDS.startedAt,
     type: FieldMetadataType.DATE_TIME,

packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts

@@ -2,6 +2,8 @@ import { Scope } from '@nestjs/common';
 
 import { isDefined } from 'twenty-shared/utils';
 
+import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum';
+import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 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';
@@ -20,8 +22,6 @@ import { getRootSteps } from 'src/modules/workflow/workflow-runner/utils/get-roo
 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';
-import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum';
-import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 
 export type RunWorkflowJobData = {
   workspaceId: string;

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts

@@ -0,0 +1 @@
+export const WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100;

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts

@@ -0,0 +1,34 @@
+import { Command, CommandRunner } from 'nest-commander';
+
+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 {
+  WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN,
+  WorkflowHandleStaledRunsJob,
+} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job';
+
+@Command({
+  name: 'cron:workflow:handle-staled-runs',
+  description: 'Handles staled workflow runs',
+})
+export class WorkflowHandleStaledRunsCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron({
+      jobName: WorkflowHandleStaledRunsJob.name,
+      data: undefined,
+      options: {
+        repeat: {
+          pattern: WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN,
+        },
+      },
+    });
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts

@@ -6,13 +6,13 @@ import { MessageQueueService } from 'src/engine/core-modules/message-queue/servi
 import {
   WORKFLOW_RUN_ENQUEUE_CRON_PATTERN,
   WorkflowRunEnqueueJob,
-} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job';
+} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job';
 
 @Command({
   name: 'cron:workflow:enqueue-awaiting-workflow-run',
   description: 'Enqueues awaiting workflow runs',
 })
-export class CronWorkflowRunEnqueueCommand extends CommandRunner {
+export class WorkflowRunEnqueueCronCommand extends CommandRunner {
   constructor(
     @InjectMessageQueue(MessageQueue.cronQueue)
     private readonly messageQueueService: MessageQueueService,

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts

@@ -0,0 +1,60 @@
+import { LessThan } from 'typeorm';
+
+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 { 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 { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
+
+export type WorkflowHandleStaledRunsPerWorkspaceJobData = {
+  workspaceId: string;
+};
+
+@Processor(MessageQueue.workflowQueue)
+export class WorkflowHandleStaledRunsPerWorkspaceJob {
+  constructor(
+    private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+  ) {}
+
+  @Process(WorkflowHandleStaledRunsPerWorkspaceJob.name)
+  async handle(data: WorkflowHandleStaledRunsPerWorkspaceJobData) {
+    const { workspaceId } = data;
+
+    const workflowRunRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace(
+        workspaceId,
+        WorkflowRunWorkspaceEntity,
+        { shouldBypassPermissionChecks: true },
+      );
+
+    const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000);
+
+    const staledWorkflowRuns = await workflowRunRepository.find({
+      where: {
+        status: WorkflowRunStatus.ENQUEUED,
+        enqueuedAt: LessThan(oneHourAgo),
+      },
+    });
+
+    if (staledWorkflowRuns.length <= 0) {
+      return;
+    }
+
+    await workflowRunRepository.update(
+      staledWorkflowRuns.map((workflowRun) => workflowRun.id),
+      {
+        enqueuedAt: null,
+        status: WorkflowRunStatus.NOT_STARTED,
+      },
+    );
+
+    await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
+      workspaceId,
+    );
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts

@@ -0,0 +1,50 @@
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+import { Repository } from 'typeorm';
+
+import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import {
+  WorkflowHandleStaledRunsPerWorkspaceJob,
+  type WorkflowHandleStaledRunsPerWorkspaceJobData,
+} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job';
+
+export const WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN = '0 * * * *';
+
+@Processor(MessageQueue.cronQueue)
+export class WorkflowHandleStaledRunsJob {
+  constructor(
+    @InjectRepository(Workspace, 'core')
+    private readonly workspaceRepository: Repository<Workspace>,
+    @InjectMessageQueue(MessageQueue.workflowQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @Process(WorkflowHandleStaledRunsJob.name)
+  @SentryCronMonitor(
+    WorkflowHandleStaledRunsJob.name,
+    WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN,
+  )
+  async handle() {
+    const activeWorkspaces = await this.workspaceRepository.find({
+      where: {
+        activationStatus: WorkspaceActivationStatus.ACTIVE,
+      },
+    });
+
+    for (const activeWorkspace of activeWorkspaces) {
+      await this.messageQueueService.add<WorkflowHandleStaledRunsPerWorkspaceJobData>(
+        WorkflowHandleStaledRunsPerWorkspaceJob.name,
+        {
+          workspaceId: activeWorkspace.id,
+        },
+      );
+    }
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts

@@ -0,0 +1,107 @@
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+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 { 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 { 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 {
+  RunWorkflowJob,
+  type RunWorkflowJobData,
+} from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job';
+import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
+
+export type WorkflowRunEnqueuePerWorkspaceJobData = {
+  workspaceId: string;
+};
+
+@Processor(MessageQueue.workflowQueue)
+export class WorkflowRunEnqueuePerWorkspaceJob {
+  constructor(
+    private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
+    @InjectMessageQueue(MessageQueue.workflowQueue)
+    private readonly messageQueueService: MessageQueueService,
+    private readonly metricsService: MetricsService,
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+  ) {}
+
+  @Process(WorkflowRunEnqueuePerWorkspaceJob.name)
+  async handle(data: WorkflowRunEnqueuePerWorkspaceJobData) {
+    const { workspaceId } = data;
+
+    try {
+      const workflowRunRepository =
+        await this.twentyORMGlobalManager.getRepositoryForWorkspace(
+          workspaceId,
+          WorkflowRunWorkspaceEntity,
+          { shouldBypassPermissionChecks: true },
+        );
+
+      const remainingWorkflowRunToEnqueueCount =
+        await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromDatabase(
+          workspaceId,
+        );
+
+      if (remainingWorkflowRunToEnqueueCount <= 0) {
+        await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
+          workspaceId,
+        );
+
+        return;
+      }
+
+      const workflowRunsToEnqueue = await workflowRunRepository.find({
+        where: {
+          status: WorkflowRunStatus.NOT_STARTED,
+        },
+        order: {
+          createdAt: 'ASC',
+        },
+        take: remainingWorkflowRunToEnqueueCount,
+      });
+
+      if (workflowRunsToEnqueue.length <= 0) {
+        await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount(
+          workspaceId,
+        );
+
+        return;
+      }
+
+      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,
+      });
+
+      throw error;
+    }
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts

@@ -1,138 +0,0 @@
-import { Logger } from '@nestjs/common';
-import { InjectRepository } from '@nestjs/typeorm';
-
-import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
-import { Repository } from 'typeorm';
-
-import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
-import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
-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 { 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 { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
-import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
-import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util';
-import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service';
-import {
-  WorkflowRunStatus,
-  WorkflowRunWorkspaceEntity,
-} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
-import {
-  RunWorkflowJob,
-  type RunWorkflowJobData,
-} from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job';
-import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
-
-export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '* * * * *';
-
-@Processor(MessageQueue.cronQueue)
-export class WorkflowRunEnqueueJob {
-  private readonly logger = new Logger(WorkflowRunEnqueueJob.name);
-
-  constructor(
-    @InjectRepository(Workspace, 'core')
-    private readonly workspaceRepository: Repository<Workspace>,
-    private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
-    private readonly workspaceDataSourceService: WorkspaceDataSourceService,
-    @InjectMessageQueue(MessageQueue.workflowQueue)
-    private readonly messageQueueService: MessageQueueService,
-    private readonly metricsService: MetricsService,
-    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
-  ) {}
-
-  @Process(WorkflowRunEnqueueJob.name)
-  @SentryCronMonitor(
-    WorkflowRunEnqueueJob.name,
-    WORKFLOW_RUN_ENQUEUE_CRON_PATTERN,
-  )
-  async handle() {
-    const activeWorkspaces = await this.workspaceRepository.find({
-      where: {
-        activationStatus: WorkspaceActivationStatus.ACTIVE,
-      },
-    });
-
-    const mainDataSource =
-      await this.workspaceDataSourceService.connectToMainDataSource();
-
-    for (const activeWorkspace of activeWorkspaces) {
-      let enqueuedWorkflowRunCount = 0;
-
-      try {
-        const remainingWorkflowRunCount =
-          await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCount(
-            activeWorkspace.id,
-          );
-
-        if (remainingWorkflowRunCount <= 0) {
-          continue;
-        }
-
-        const schemaName = getWorkspaceSchemaName(activeWorkspace.id);
-
-        // Using raw query to avoid storing repository in cache
-        const workflowRuns = await mainDataSource.query(
-          `SELECT id FROM ${schemaName}."workflowRun" WHERE status = '${WorkflowRunStatus.NOT_STARTED}' ORDER BY "createdAt" ASC`,
-        );
-
-        const workflowRunsToEnqueueCount = Math.min(
-          remainingWorkflowRunCount,
-          workflowRuns.length,
-        );
-
-        if (workflowRunsToEnqueueCount <= 0) {
-          continue;
-        }
-
-        const workflowRunRepository =
-          await this.twentyORMGlobalManager.getRepositoryForWorkspace(
-            activeWorkspace.id,
-            WorkflowRunWorkspaceEntity,
-            { shouldBypassPermissionChecks: true },
-          );
-
-        for (
-          let runIndex = 0;
-          runIndex < workflowRunsToEnqueueCount;
-          runIndex++
-        ) {
-          const workflowRunId = workflowRuns[runIndex].id;
-
-          await this.messageQueueService.add<RunWorkflowJobData>(
-            RunWorkflowJob.name,
-            {
-              workflowRunId,
-              workspaceId: activeWorkspace.id,
-            },
-          );
-
-          await workflowRunRepository.update(workflowRunId, {
-            status: WorkflowRunStatus.ENQUEUED,
-          });
-
-          enqueuedWorkflowRunCount++;
-        }
-      } catch (error) {
-        this.logger.error(
-          `Error enqueuing workflow runs for workspace ${activeWorkspace.id}`,
-          error,
-        );
-
-        this.metricsService.incrementCounter({
-          key: MetricsKeys.WorkflowRunFailedToEnqueue,
-          eventId: activeWorkspace.id,
-        });
-      } finally {
-        if (enqueuedWorkflowRunCount > 0) {
-          await this.workflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount(
-            activeWorkspace.id,
-            enqueuedWorkflowRunCount,
-          );
-        }
-      }
-    }
-  }
-}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts

@@ -0,0 +1,50 @@
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+import { Repository } from 'typeorm';
+
+import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import {
+  WorkflowRunEnqueuePerWorkspaceJob,
+  type WorkflowRunEnqueuePerWorkspaceJobData,
+} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job';
+
+export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '* * * * *';
+
+@Processor(MessageQueue.cronQueue)
+export class WorkflowRunEnqueueJob {
+  constructor(
+    @InjectRepository(Workspace, 'core')
+    private readonly workspaceRepository: Repository<Workspace>,
+    @InjectMessageQueue(MessageQueue.workflowQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @Process(WorkflowRunEnqueueJob.name)
+  @SentryCronMonitor(
+    WorkflowRunEnqueueJob.name,
+    WORKFLOW_RUN_ENQUEUE_CRON_PATTERN,
+  )
+  async handle() {
+    const activeWorkspaces = await this.workspaceRepository.find({
+      where: {
+        activationStatus: WorkspaceActivationStatus.ACTIVE,
+      },
+    });
+
+    for (const activeWorkspace of activeWorkspaces) {
+      await this.messageQueueService.add<WorkflowRunEnqueuePerWorkspaceJobData>(
+        WorkflowRunEnqueuePerWorkspaceJob.name,
+        {
+          workspaceId: activeWorkspace.id,
+        },
+      );
+    }
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts

@@ -6,11 +6,15 @@ import { MessageQueueModule } from 'src/engine/core-modules/message-queue/messag
 import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
 import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
-import { CronWorkflowRunEnqueueCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-workflow-run-enqueue.cron.command';
-import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job';
-import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
-import { CleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job';
 import { CronCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-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 { CleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job';
+import { WorkflowHandleStaledRunsPerWorkspaceJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job';
+import { WorkflowHandleStaledRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job';
+import { WorkflowRunEnqueuePerWorkspaceJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job';
+import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job';
+import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
 
 @Module({
   imports: [
@@ -22,11 +26,19 @@ import { CronCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runn
   ],
   providers: [
     WorkflowRunQueueWorkspaceService,
+    WorkflowRunEnqueueCronCommand,
     WorkflowRunEnqueueJob,
+    WorkflowRunEnqueuePerWorkspaceJob,
+    WorkflowHandleStaledRunsCronCommand,
+    WorkflowHandleStaledRunsJob,
+    WorkflowHandleStaledRunsPerWorkspaceJob,
     CleanWorkflowRunsJob,
-    CronWorkflowRunEnqueueCommand,
     CronCleanWorkflowRunsCommand,
   ],
-  exports: [WorkflowRunQueueWorkspaceService],
+  exports: [
+    WorkflowRunQueueWorkspaceService,
+    WorkflowRunEnqueueCronCommand,
+    WorkflowHandleStaledRunsCronCommand,
+  ],
 })
 export class WorkflowRunQueueModule {}

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts

@@ -1,17 +1,24 @@
 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 { WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit';
 import { getWorkflowRunQueuedCountCacheKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-count-key.util';
 
 @Injectable()
 export class WorkflowRunQueueWorkspaceService {
-  private readonly WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100;
-
   constructor(
     @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow)
     private readonly cacheStorage: CacheStorageService,
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
   ) {}
 
   async increaseWorkflowRunQueuedCount(
@@ -40,11 +47,64 @@ export class WorkflowRunQueueWorkspaceService {
     );
   }
 
-  async getRemainingRunsToEnqueueCount(workspaceId: string): Promise<number> {
+  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);
 
-    return this.WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount;
+    return WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - 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]),
+      },
+    });
+
+    return WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount;
+  }
+
+  private async setWorkflowRunQueuedCount(
+    workspaceId: string,
+    count: number,
+  ): Promise<void> {
+    await this.cacheStorage.set(
+      getWorkflowRunQueuedCountCacheKey(workspaceId),
+      count,
+    );
   }
 
   private async getCurrentWorkflowRunQueuedCount(

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts

@@ -1,7 +1,7 @@
 import { Injectable } from '@nestjs/common';
 
 import { isDefined } from 'twenty-shared/utils';
-import { StepStatus, WorkflowRunStepInfo } from 'twenty-shared/workflow';
+import { StepStatus, type WorkflowRunStepInfo } from 'twenty-shared/workflow';
 import { type QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity';
 import { v4 } from 'uuid';
 
@@ -44,9 +44,9 @@ export class WorkflowRunWorkspaceService {
   }: {
     workflowVersionId: string;
     createdBy: ActorMetadata;
-    workflowRunId?: string;
     status: WorkflowRunStatus.NOT_STARTED | WorkflowRunStatus.ENQUEUED;
     triggerPayload: object;
+    workflowRunId?: string;
   }) {
     const workspaceId =
       this.scopedWorkspaceContextFactory.create()?.workspaceId;
@@ -124,6 +124,8 @@ export class WorkflowRunWorkspaceService {
       status,
       position,
       state: initState,
+      enqueuedAt:
+        status === WorkflowRunStatus.ENQUEUED ? new Date().toISOString() : null,
     });
 
     await workflowRunRepository.insert(workflowRun);
@@ -144,14 +146,21 @@ export class WorkflowRunWorkspaceService {
       workspaceId,
     });
 
+    if (
+      workflowRunToUpdate.status === WorkflowRunStatus.COMPLETED ||
+      workflowRunToUpdate.status === WorkflowRunStatus.FAILED
+    ) {
+      throw new WorkflowRunException(
+        'Cannot start a workflow run already ended',
+        WorkflowRunExceptionCode.INVALID_OPERATION,
+      );
+    }
+
     if (
       workflowRunToUpdate.status !== WorkflowRunStatus.ENQUEUED &&
       workflowRunToUpdate.status !== WorkflowRunStatus.NOT_STARTED
     ) {
-      throw new WorkflowRunException(
-        'Workflow run already started',
-        WorkflowRunExceptionCode.INVALID_OPERATION,
-      );
+      return;
     }
 
     const partialUpdate = {

packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts

@@ -9,7 +9,7 @@ import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/
 import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
 import {
   RunWorkflowJob,
-  type RunWorkflowJobData,
+  RunWorkflowJobData,
 } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job';
 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';
@@ -56,7 +56,7 @@ export class WorkflowRunnerWorkspaceService {
       });
 
     const remainingRunsToEnqueueCount =
-      await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCount(
+      await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromCache(
         workspaceId,
       );
 

packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts

@@ -2,12 +2,12 @@ import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';
 
 import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
 import { AutomatedTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service';
-import { CronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command';
+import { WorkflowCronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command';
 import { CronTriggerCronJob } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/cron-trigger.cron.job';
 import { DatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener';
-import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 
 @Module({
   imports: [
@@ -19,8 +19,8 @@ import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/works
     AutomatedTriggerWorkspaceService,
     DatabaseEventTriggerListener,
     CronTriggerCronJob,
-    CronTriggerCronCommand,
+    WorkflowCronTriggerCronCommand,
   ],
-  exports: [AutomatedTriggerWorkspaceService, CronTriggerCronCommand],
+  exports: [AutomatedTriggerWorkspaceService, WorkflowCronTriggerCronCommand],
 })
 export class AutomatedTriggerModule {}

packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command.ts

@@ -12,7 +12,7 @@ import {
   name: 'cron:workflow:automated-cron-trigger',
   description: 'Starts a cron job to trigger cron triggered workflows',
 })
-export class CronTriggerCronCommand extends CommandRunner {
+export class WorkflowCronTriggerCronCommand extends CommandRunner {
   constructor(
     @InjectMessageQueue(MessageQueue.cronQueue)
     private readonly messageQueueService: MessageQueueService,