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 e98f279cdf..199ed4b4a3 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 @@ -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}`, + ); + }, + ); + } } 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 e2182ce256..cd65d29433 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,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, @@ -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; + } } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts index 78bbc7a5aa..8275127139 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts +++ b/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, + ); + }, + ); + } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts index 873fcc9f53..5a56e06ce8 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts +++ b/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,