Add logs to debug workflow crons (#17302)

We have 3 crons that get a lot of timeout in sentry. 
Adding logs to help debugging.

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

@@ -39,54 +39,79 @@ export class WorkflowCleanWorkflowRunsJob {
     CLEAN_WORKFLOW_RUN_CRON_PATTERN,
   )
   async handle() {
-    const activeWorkspaces = await this.workspaceRepository.find({
-      where: {
-        activationStatus: WorkspaceActivationStatus.ACTIVE,
-      },
-    });
+    this.logger.log('Starting WorkflowCleanWorkflowRunsJob cron');
 
-    for (const activeWorkspace of activeWorkspaces) {
-      const schemaName = getWorkspaceSchemaName(activeWorkspace.id);
-
-      const authContext = buildSystemAuthContext(activeWorkspace.id);
-
-      await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
-        authContext,
-        async () => {
-          const workflowRunsToDelete = await this.coreDataSource.query(
-            `
-            WITH ranked_runs AS (
-              SELECT id,
-                     ROW_NUMBER() OVER (
-                        PARTITION BY "workflowId"
-                        ORDER BY "createdAt" DESC
-                     ) AS rn,
-                     "createdAt"
-              FROM ${schemaName}."workflowRun"
-              WHERE status IN ('${WorkflowRunStatus.COMPLETED}', '${WorkflowRunStatus.FAILED}')
-            )
-            SELECT id, rn FROM ranked_runs
-            WHERE rn > ${NUMBER_OF_WORKFLOW_RUNS_TO_KEEP}
-               OR "createdAt" < NOW() - INTERVAL '14 days';
-          `,
-          );
-
-          const workflowRunRepository =
-            await this.globalWorkspaceOrmManager.getRepository(
-              activeWorkspace.id,
-              WorkflowRunWorkspaceEntity,
-              { shouldBypassPermissionChecks: true },
-            );
-
-          for (const workflowRunToDelete of workflowRunsToDelete) {
-            await workflowRunRepository.delete(workflowRunToDelete.id);
-          }
-
-          this.logger.log(
-            `Deleted ${workflowRunsToDelete.length} workflow runs for workspace ${activeWorkspace.id} (schema ${schemaName})`,
-          );
+    try {
+      const activeWorkspaces = await this.workspaceRepository.find({
+        where: {
+          activationStatus: WorkspaceActivationStatus.ACTIVE,
         },
-      );
+      });
+
+      for (let i = 0; i < activeWorkspaces.length; i++) {
+        const activeWorkspace = activeWorkspaces[i];
+
+        this.logger.log(
+          `Processing workspace ${activeWorkspace.id} (${i + 1}/${activeWorkspaces.length})`,
+        );
+
+        try {
+          await this.cleanWorkflowRunsForWorkspace(activeWorkspace.id);
+        } catch (error) {
+          this.logger.error(
+            `Failed to clean workflow runs for workspace ${activeWorkspace.id}`,
+            error,
+          );
+        }
+      }
+
+      this.logger.log('Completed WorkflowCleanWorkflowRunsJob cron');
+    } catch (error) {
+      this.logger.error('WorkflowCleanWorkflowRunsJob cron failed', error);
+      throw error;
     }
   }
+
+  private async cleanWorkflowRunsForWorkspace(workspaceId: string) {
+    const schemaName = getWorkspaceSchemaName(workspaceId);
+    const authContext = buildSystemAuthContext(workspaceId);
+
+    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
+      authContext,
+      async () => {
+        const workflowRunsToDelete = await this.coreDataSource.query(
+          `
+          WITH ranked_runs AS (
+            SELECT id,
+                   ROW_NUMBER() OVER (
+                      PARTITION BY "workflowId"
+                      ORDER BY "createdAt" DESC
+                   ) AS rn,
+                   "createdAt"
+            FROM ${schemaName}."workflowRun"
+            WHERE status IN ('${WorkflowRunStatus.COMPLETED}', '${WorkflowRunStatus.FAILED}')
+          )
+          SELECT id, rn FROM ranked_runs
+          WHERE rn > ${NUMBER_OF_WORKFLOW_RUNS_TO_KEEP}
+             OR "createdAt" < NOW() - INTERVAL '14 days';
+        `,
+        );
+
+        const workflowRunRepository =
+          await this.globalWorkspaceOrmManager.getRepository(
+            workspaceId,
+            WorkflowRunWorkspaceEntity,
+            { shouldBypassPermissionChecks: true },
+          );
+
+        for (const workflowRunToDelete of workflowRunsToDelete) {
+          await workflowRunRepository.delete(workflowRunToDelete.id);
+        }
+
+        this.logger.log(
+          `Deleted ${workflowRunsToDelete.length} workflow runs for workspace ${workspaceId}`,
+        );
+      },
+    );
+  }
 }

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

@@ -1,3 +1,4 @@
+import { Logger } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
@@ -14,6 +15,8 @@ export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '*/5 * * * *';
 
 @Processor(MessageQueue.cronQueue)
 export class WorkflowRunEnqueueCronJob {
+  private readonly logger = new Logger(WorkflowRunEnqueueCronJob.name);
+
   constructor(
     @InjectRepository(WorkspaceEntity)
     private readonly workspaceRepository: Repository<WorkspaceEntity>,
@@ -26,15 +29,41 @@ export class WorkflowRunEnqueueCronJob {
     WORKFLOW_RUN_ENQUEUE_CRON_PATTERN,
   )
   async handle() {
-    const activeWorkspaces = await this.workspaceRepository.find({
-      where: {
-        activationStatus: WorkspaceActivationStatus.ACTIVE,
-      },
-    });
+    this.logger.log('Starting WorkflowRunEnqueueCronJob cron');
 
-    await this.workflowRunEnqueueWorkspaceService.enqueueRuns({
-      workspaceIds: activeWorkspaces.map((workspace) => workspace.id),
-      isCacheMode: false,
-    });
+    try {
+      const activeWorkspaces = await this.workspaceRepository.find({
+        where: {
+          activationStatus: WorkspaceActivationStatus.ACTIVE,
+        },
+      });
+
+      for (let i = 0; i < activeWorkspaces.length; i++) {
+        const workspace = activeWorkspaces[i];
+
+        this.logger.log(
+          `Processing workspace ${workspace.id} (${i + 1}/${activeWorkspaces.length})`,
+        );
+
+        try {
+          await this.workflowRunEnqueueWorkspaceService.enqueueRunsForWorkspace(
+            {
+              workspaceId: workspace.id,
+              isCacheMode: false,
+            },
+          );
+        } catch (error) {
+          this.logger.error(
+            `Failed to enqueue runs for workspace ${workspace.id}`,
+            error,
+          );
+        }
+      }
+
+      this.logger.log('Completed WorkflowRunEnqueueCronJob cron');
+    } catch (error) {
+      this.logger.error('WorkflowRunEnqueueCronJob cron failed', error);
+      throw error;
+    }
   }
 }

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

@@ -21,52 +21,71 @@ export class WorkflowHandleStaledRunsWorkspaceService {
   ) {}
 
   async handleStaledRuns({ workspaceIds }: { workspaceIds: string[] }) {
-    for (const workspaceId of workspaceIds) {
-      try {
-        const authContext = buildSystemAuthContext(workspaceId);
+    this.logger.log('Starting handleStaledRuns');
 
-        await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
-          authContext,
-          async () => {
-            const workflowRunRepository =
-              await this.globalWorkspaceOrmManager.getRepository(
-                workspaceId,
-                WorkflowRunWorkspaceEntity,
-                { shouldBypassPermissionChecks: true },
-              );
+    try {
+      for (let i = 0; i < workspaceIds.length; i++) {
+        const workspaceId = workspaceIds[i];
 
-            const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000);
-
-            const staledWorkflowRuns = await workflowRunRepository.find({
-              where: {
-                status: WorkflowRunStatus.ENQUEUED,
-                enqueuedAt: Or(LessThan(oneHourAgo), IsNull()),
-              },
-            });
-
-            if (staledWorkflowRuns.length <= 0) {
-              return;
-            }
-
-            await workflowRunRepository.update(
-              staledWorkflowRuns.map((workflowRun) => workflowRun.id),
-              {
-                enqueuedAt: null,
-                status: WorkflowRunStatus.NOT_STARTED,
-              },
-            );
-
-            await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(
-              workspaceId,
-            );
-          },
-        );
-      } catch (error) {
-        this.logger.error(
-          `Failed to handle staled runs for workspace: ${workspaceId}`,
-          error,
+        this.logger.log(
+          `Processing workspace ${workspaceId} (${i + 1}/${workspaceIds.length})`,
         );
+
+        try {
+          await this.handleStaledRunsForWorkspace(workspaceId);
+        } catch (error) {
+          this.logger.error(
+            `Failed to handle staled runs for workspace ${workspaceId}`,
+            error,
+          );
+        }
       }
+
+      this.logger.log('Completed handleStaledRuns');
+    } catch (error) {
+      this.logger.error('handleStaledRuns failed', error);
+      throw error;
     }
   }
+
+  private async handleStaledRunsForWorkspace(workspaceId: string) {
+    const authContext = buildSystemAuthContext(workspaceId);
+
+    await this.globalWorkspaceOrmManager.executeInWorkspaceContext(
+      authContext,
+      async () => {
+        const workflowRunRepository =
+          await this.globalWorkspaceOrmManager.getRepository(
+            workspaceId,
+            WorkflowRunWorkspaceEntity,
+            { shouldBypassPermissionChecks: true },
+          );
+
+        const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000);
+
+        const staledWorkflowRuns = await workflowRunRepository.find({
+          where: {
+            status: WorkflowRunStatus.ENQUEUED,
+            enqueuedAt: Or(LessThan(oneHourAgo), IsNull()),
+          },
+        });
+
+        if (staledWorkflowRuns.length <= 0) {
+          return;
+        }
+
+        await workflowRunRepository.update(
+          staledWorkflowRuns.map((workflowRun) => workflowRun.id),
+          {
+            enqueuedAt: null,
+            status: WorkflowRunStatus.NOT_STARTED,
+          },
+        );
+
+        await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(
+          workspaceId,
+        );
+      },
+    );
+  }
 }

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

@@ -28,18 +28,6 @@ export class WorkflowRunEnqueueWorkspaceService {
     private readonly metricsService: MetricsService,
   ) {}
 
-  async enqueueRuns({
-    workspaceIds,
-    isCacheMode,
-  }: {
-    workspaceIds: string[];
-    isCacheMode: boolean;
-  }) {
-    for (const workspaceId of workspaceIds) {
-      await this.enqueueRunsForWorkspace({ workspaceId, isCacheMode });
-    }
-  }
-
   async enqueueRunsForWorkspace({
     workspaceId,
     isCacheMode,