Fix staled run cron (#14044)

- Dequeue workflows when enqueuedAt is Null
- Rename cron cleaner + add it to the registered crons

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

@@ -10,6 +10,7 @@ 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 { WorkflowCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-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 { WorkflowCronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command';
@@ -33,6 +34,7 @@ export class CronRegisterAllCommand extends CommandRunner {
     private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand,
     private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand,
     private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand,
+    private readonly workflowCleanWorkflowRunsCronCommand: WorkflowCleanWorkflowRunsCommand,
   ) {
     super();
   }
@@ -85,6 +87,10 @@ export class CronRegisterAllCommand extends CommandRunner {
         name: 'WorkflowHandleStaledRuns',
         command: this.workflowHandleStaledRunsCronCommand,
       },
+      {
+        name: 'WorkflowCleanWorkflowRuns',
+        command: this.workflowCleanWorkflowRunsCronCommand,
+      },
     ];
 
     let successCount = 0;

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

@@ -5,14 +5,14 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu
 import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import {
   CLEAN_WORKFLOW_RUN_CRON_PATTERN,
-  CleanWorkflowRunsJob,
+  WorkflowCleanWorkflowRunsJob,
-} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job';
+} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job';
 
 @Command({
   name: 'cron:workflow:clean-workflow-runs',
   description: 'Clean workflow runs',
 })
-export class CronCleanWorkflowRunsCommand extends CommandRunner {
+export class WorkflowCleanWorkflowRunsCommand extends CommandRunner {
   constructor(
     @InjectMessageQueue(MessageQueue.cronQueue)
     private readonly messageQueueService: MessageQueueService,
@@ -22,7 +22,7 @@ export class CronCleanWorkflowRunsCommand extends CommandRunner {
 
   async run(): Promise<void> {
     await this.messageQueueService.addCron({
-      jobName: CleanWorkflowRunsJob.name,
+      jobName: WorkflowCleanWorkflowRunsJob.name,
       data: undefined,
       options: {
         repeat: {

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

@@ -22,8 +22,8 @@ export const CLEAN_WORKFLOW_RUN_CRON_PATTERN = '0 0 * * *';
 const NUMBER_OF_WORKFLOW_RUNS_TO_KEEP = 1000;
 
 @Processor(MessageQueue.cronQueue)
-export class CleanWorkflowRunsJob {
+export class WorkflowCleanWorkflowRunsJob {
-  private readonly logger = new Logger(CleanWorkflowRunsJob.name);
+  private readonly logger = new Logger(WorkflowCleanWorkflowRunsJob.name);
 
   constructor(
     @InjectRepository(Workspace, 'core')
@@ -32,8 +32,11 @@ export class CleanWorkflowRunsJob {
     private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
   ) {}
 
-  @Process(CleanWorkflowRunsJob.name)
+  @Process(WorkflowCleanWorkflowRunsJob.name)
-  @SentryCronMonitor(CleanWorkflowRunsJob.name, CLEAN_WORKFLOW_RUN_CRON_PATTERN)
+  @SentryCronMonitor(
+    WorkflowCleanWorkflowRunsJob.name,
+    CLEAN_WORKFLOW_RUN_CRON_PATTERN,
+  )
   async handle() {
     const activeWorkspaces = await this.workspaceRepository.find({
       where: {

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

@@ -1,4 +1,4 @@
-import { LessThan } from 'typeorm';
+import { IsNull, LessThan, Or } 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';
@@ -37,7 +37,7 @@ export class WorkflowHandleStaledRunsPerWorkspaceJob {
     const staledWorkflowRuns = await workflowRunRepository.find({
       where: {
         status: WorkflowRunStatus.ENQUEUED,
-        enqueuedAt: LessThan(oneHourAgo),
+        enqueuedAt: Or(LessThan(oneHourAgo), IsNull()),
       },
     });
 

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

@@ -6,10 +6,10 @@ 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 { CronCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-clean-workflow-runs.cron.command';
+import { WorkflowCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-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 { WorkflowCleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-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';
@@ -32,13 +32,14 @@ import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-
     WorkflowHandleStaledRunsCronCommand,
     WorkflowHandleStaledRunsJob,
     WorkflowHandleStaledRunsPerWorkspaceJob,
-    CleanWorkflowRunsJob,
+    WorkflowCleanWorkflowRunsJob,
-    CronCleanWorkflowRunsCommand,
+    WorkflowCleanWorkflowRunsCommand,
   ],
   exports: [
     WorkflowRunQueueWorkspaceService,
     WorkflowRunEnqueueCronCommand,
     WorkflowHandleStaledRunsCronCommand,
+    WorkflowCleanWorkflowRunsCommand,
   ],
 })
 export class WorkflowRunQueueModule {}