From 911a46aa45505478bf6e2fbc61f8494ec8e1dbd7 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Wed, 4 Mar 2026 18:29:12 +0100 Subject: [PATCH] Improve workflow perfs (#18376) Workflow crons take a few minutes to run. Loading each repo takes ~200 to 300ms locally. Adding a lite mode so it takes less than 100ms. Also doing batch promises. Finally, cleaning runs timeout when there are too many. Doing batches as well. --- .../global-workspace-orm.manager.ts | 53 +++++++- ...ecute-in-workspace-context-options.type.ts | 3 + .../iterator-action.workflow-action.spec.ts | 12 +- .../iterator/iterator.workflow-action.ts | 11 +- .../workflow-executor.workspace-service.ts | 64 ++++----- .../workflow-runner/jobs/run-workflow.job.ts | 37 +++++- .../workflow-clean-workflow-runs.cron.job.ts | 55 +++++--- .../workflow-handle-staled-runs.cron.job.ts | 55 +++++--- .../jobs/workflow-run-enqueue.cron.job.ts | 56 +++++--- .../jobs/workflow-clean-workflow-runs.job.ts | 124 ++++++++++++++---- .../workflow-run.workspace-service.ts | 16 ++- 11 files changed, 354 insertions(+), 132 deletions(-) create mode 100644 packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/types/execute-in-workspace-context-options.type.ts diff --git a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager.ts b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager.ts index ff36091fb5..d9ff786192 100644 --- a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager.ts +++ b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager.ts @@ -2,11 +2,12 @@ import { Injectable, type Type } from '@nestjs/common'; import { type ObjectLiteral } from 'typeorm'; -import { type WorkspaceAuthContext } from 'src/engine/core-modules/auth/types/workspace-auth-context.type'; import { getWorkspaceAuthContext } from 'src/engine/core-modules/auth/storage/workspace-auth-context.storage'; +import { type WorkspaceAuthContext } from 'src/engine/core-modules/auth/types/workspace-auth-context.type'; import { buildObjectIdByNameMaps } from 'src/engine/metadata-modules/flat-object-metadata/utils/build-object-id-by-name-maps.util'; import { GlobalWorkspaceDataSource } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource'; import { GlobalWorkspaceDataSourceService } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.service'; +import { ExecuteInWorkspaceContextOptions } from 'src/engine/twenty-orm/global-workspace-datasource/types/execute-in-workspace-context-options.type'; import type { WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository'; import { type ORMWorkspaceContext, @@ -69,9 +70,12 @@ export class GlobalWorkspaceOrmManager { async executeInWorkspaceContext( fn: () => T | Promise, authContext?: WorkspaceAuthContext, + options?: ExecuteInWorkspaceContextOptions, ): Promise { const resolvedAuthContext = authContext ?? getWorkspaceAuthContext(); - const context = await this.loadWorkspaceContext(resolvedAuthContext); + const context = options?.lite + ? await this.loadLiteWorkspaceContext(resolvedAuthContext) + : await this.loadWorkspaceContext(resolvedAuthContext); return withWorkspaceContext(context, fn); } @@ -120,4 +124,49 @@ export class GlobalWorkspaceOrmManager { userWorkspaceRoleMap, }; } + + private async loadLiteWorkspaceContext( + authContext: WorkspaceAuthContext, + ): Promise { + const workspaceId = authContext.workspace.id; + + const { + flatObjectMetadataMaps, + flatFieldMetadataMaps, + ORMEntityMetadatas: entityMetadatas, + } = await this.workspaceCacheService.getOrRecompute(workspaceId, [ + 'flatObjectMetadataMaps', + 'flatFieldMetadataMaps', + 'ORMEntityMetadatas', + ]); + + const { idByNameSingular: objectIdByNameSingular } = + buildObjectIdByNameMaps(flatObjectMetadataMaps); + + return { + authContext, + flatObjectMetadataMaps, + flatFieldMetadataMaps, + flatIndexMaps: { + byUniversalIdentifier: {}, + universalIdentifierById: {}, + universalIdentifiersByApplicationId: {}, + }, + flatRowLevelPermissionPredicateMaps: { + byUniversalIdentifier: {}, + universalIdentifierById: {}, + universalIdentifiersByApplicationId: {}, + }, + flatRowLevelPermissionPredicateGroupMaps: { + byUniversalIdentifier: {}, + universalIdentifierById: {}, + universalIdentifiersByApplicationId: {}, + }, + objectIdByNameSingular, + featureFlagsMap: {} as ORMWorkspaceContext['featureFlagsMap'], + permissionsPerRoleId: {}, + entityMetadatas, + userWorkspaceRoleMap: {}, + }; + } } diff --git a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/types/execute-in-workspace-context-options.type.ts b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/types/execute-in-workspace-context-options.type.ts new file mode 100644 index 0000000000..0f5fb86c63 --- /dev/null +++ b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/types/execute-in-workspace-context-options.type.ts @@ -0,0 +1,3 @@ +export type ExecuteInWorkspaceContextOptions = { + lite?: boolean; +}; diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/__tests__/iterator-action.workflow-action.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/__tests__/iterator-action.workflow-action.spec.ts index 442b2e96b3..4e38237b4c 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/__tests__/iterator-action.workflow-action.spec.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/__tests__/iterator-action.workflow-action.spec.ts @@ -192,9 +192,9 @@ describe('IteratorWorkflowAction', () => { }, } as any; - workflowRunWorkspaceService.getWorkflowRunOrFail - .mockResolvedValueOnce(mockStepInfo) - .mockResolvedValueOnce(mockStepInfo); + workflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + mockStepInfo, + ); const result = await service.execute(input); @@ -243,9 +243,9 @@ describe('IteratorWorkflowAction', () => { }, } as any; - workflowRunWorkspaceService.getWorkflowRunOrFail - .mockResolvedValueOnce(mockStepInfo) - .mockResolvedValueOnce(mockStepInfo); + workflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + mockStepInfo, + ); const result = await service.execute(input); diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/iterator.workflow-action.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/iterator.workflow-action.ts index cd7ca8fe2f..8a49c86d95 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/iterator.workflow-action.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/iterator.workflow-action.ts @@ -113,6 +113,7 @@ export class IteratorWorkflowAction implements WorkflowActionInterface { workflowRunId: runInfo.workflowRunId, workspaceId: runInfo.workspaceId, steps, + stepInfos, }); } @@ -129,6 +130,7 @@ export class IteratorWorkflowAction implements WorkflowActionInterface { workflowRunId, workspaceId, steps, + stepInfos, }: { iteratorStepId: string; initialLoopStepIds: string[]; @@ -136,17 +138,10 @@ export class IteratorWorkflowAction implements WorkflowActionInterface { workflowRunId: string; workspaceId: string; steps: WorkflowAction[]; + stepInfos: Record; }) { let stepInfosToUpdate: Record = {}; - const workflowRunToUpdate = - await this.workflowRunWorkspaceService.getWorkflowRunOrFail({ - workflowRunId, - workspaceId, - }); - - const stepInfos = workflowRunToUpdate.state.stepInfos; - if (!hasProcessedAllItems) { const subStepsInfos = await this.buildSubStepInfosReset({ iteratorStepId, diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts index 75895e2597..6d94d12d61 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts @@ -490,7 +490,7 @@ export class WorkflowExecutorWorkspaceService { } } - private async skipAndFailSafelyStepsThenContinue({ + async skipAndFailSafelyStepsThenContinue({ stepIdsToSkip, stepIdsToFailSafely, steps, @@ -505,39 +505,41 @@ export class WorkflowExecutorWorkspaceService { workspaceId: string; executedStepsCount: number; }) { - const stepsToSkip = stepIdsToSkip.map((stepId) => ({ - stepId, - status: StepStatus.SKIPPED, - })); - const stepsToFailSafely = stepIdsToFailSafely.map((stepId) => ({ - stepId, - status: StepStatus.FAILED_SAFELY, - })); - const stepsToProcess = [...stepsToSkip, ...stepsToFailSafely]; + const stepInfos: Record = {}; - await Promise.all( - stepsToProcess.map(async ({ stepId, status }) => { - await this.workflowRunWorkspaceService.updateWorkflowRunStepInfo({ - stepId, - stepInfo: { status }, - workflowRunId, - workspaceId, - }); + for (const stepId of stepIdsToSkip) { + stepInfos[stepId] = { status: StepStatus.SKIPPED }; + } - const step = steps.find((step) => step.id === stepId); - const stepNextStepIds = step?.nextStepIds ?? []; + for (const stepId of stepIdsToFailSafely) { + stepInfos[stepId] = { status: StepStatus.FAILED_SAFELY }; + } - if (stepNextStepIds.length > 0) { - await this.executeFromSteps({ - stepIds: stepNextStepIds, - workflowRunId, - workspaceId, - shouldComputeWorkflowRunStatus: false, - executedStepsCount, - }); - } - }), - ); + await this.workflowRunWorkspaceService.updateWorkflowRunStepInfos({ + stepInfos, + workflowRunId, + workspaceId, + }); + + const nextStepIds = new Set(); + + for (const stepId of [...stepIdsToSkip, ...stepIdsToFailSafely]) { + const step = steps.find((step) => step.id === stepId); + + for (const nextStepId of step?.nextStepIds ?? []) { + nextStepIds.add(nextStepId); + } + } + + if (nextStepIds.size > 0) { + await this.executeFromSteps({ + stepIds: Array.from(nextStepIds), + workflowRunId, + workspaceId, + shouldComputeWorkflowRunStatus: false, + executedStepsCount, + }); + } } private async continueExecutionFromStepInAnotherJob({ diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts index 0c249542d8..a91d83ad37 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts @@ -156,13 +156,19 @@ export class RunWorkflowJob { const lastExecutedStepOutput = workflowRun.state?.stepInfos[lastExecutedStepId]; - const { nextStepIdsToExecute } = + const { nextStepIdsToExecute, nextStepIdsToSkip, nextStepIdsToFailSafely } = await this.workflowExecutorWorkspaceService.getNextStepIdsToExecute({ executedStep: lastExecutedStep, executedStepOutput: lastExecutedStepOutput, }); - if (!isDefined(nextStepIdsToExecute) || nextStepIdsToExecute.length === 0) { + const hasStepsToSkipOrFailSafely = + isDefined(nextStepIdsToSkip) || isDefined(nextStepIdsToFailSafely); + + const hasStepsToExecute = + isDefined(nextStepIdsToExecute) && nextStepIdsToExecute.length > 0; + + if (!hasStepsToSkipOrFailSafely && !hasStepsToExecute) { await this.workflowRunWorkspaceService.endWorkflowRun({ workflowRunId, workspaceId, @@ -172,11 +178,28 @@ export class RunWorkflowJob { return; } - await this.workflowExecutorWorkspaceService.executeFromSteps({ - stepIds: nextStepIdsToExecute, - workflowRunId, - workspaceId, - }); + const steps = workflowRun.state?.flow?.steps ?? []; + + if (hasStepsToSkipOrFailSafely) { + await this.workflowExecutorWorkspaceService.skipAndFailSafelyStepsThenContinue( + { + stepIdsToSkip: nextStepIdsToSkip ?? [], + stepIdsToFailSafely: nextStepIdsToFailSafely ?? [], + steps, + workflowRunId, + workspaceId, + executedStepsCount: 0, + }, + ); + } + + if (hasStepsToExecute) { + await this.workflowExecutorWorkspaceService.executeFromSteps({ + stepIds: nextStepIdsToExecute, + workflowRunId, + workspaceId, + }); + } } private async incrementTriggerMetrics({ 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 c37b34614d..d4e893268f 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 @@ -27,6 +27,8 @@ import { getRunsToCleanFindOptions } from 'src/modules/workflow/workflow-runner/ export const CLEAN_WORKFLOW_RUN_CRON_PATTERN = '0 0 * * *'; +const WORKSPACE_BATCH_SIZE = 50; + @Processor(MessageQueue.cronQueue) export class WorkflowCleanWorkflowRunsCronJob { private readonly logger = new Logger(WorkflowCleanWorkflowRunsCronJob.name); @@ -57,25 +59,30 @@ export class WorkflowCleanWorkflowRunsCronJob { let enqueuedCount = 0; - for (const workspace of activeWorkspaces) { - try { - const hasRunsToClean = await this.hasRunsToClean(workspace.id); + for ( + let workspaceIndex = 0; + workspaceIndex < activeWorkspaces.length; + workspaceIndex += WORKSPACE_BATCH_SIZE + ) { + const batch = activeWorkspaces.slice( + workspaceIndex, + workspaceIndex + WORKSPACE_BATCH_SIZE, + ); - if (hasRunsToClean) { - await this.messageQueueService.add( - WorkflowCleanWorkflowRunsJob.name, - { - workspaceId: workspace.id, - }, - ); + const results = await Promise.allSettled( + batch.map((workspace) => this.checkAndEnqueue(workspace.id)), + ); + + for (const [index, result] of results.entries()) { + if (result.status === 'fulfilled' && result.value) { enqueuedCount++; } - } catch (error) { - this.exceptionHandlerService.captureExceptions([error], { - workspace: { - id: workspace.id, - }, - }); + + if (result.status === 'rejected') { + this.exceptionHandlerService.captureExceptions([result.reason], { + workspace: { id: batch[index].id }, + }); + } } } @@ -84,6 +91,21 @@ export class WorkflowCleanWorkflowRunsCronJob { ); } + private async checkAndEnqueue(workspaceId: string): Promise { + const hasRunsToClean = await this.hasRunsToClean(workspaceId); + + if (hasRunsToClean) { + await this.messageQueueService.add( + WorkflowCleanWorkflowRunsJob.name, + { workspaceId }, + ); + + return true; + } + + return false; + } + private async hasRunsToClean(workspaceId: string): Promise { const authContext = buildSystemAuthContext(workspaceId); @@ -113,6 +135,7 @@ export class WorkflowCleanWorkflowRunsCronJob { return totalCompletedRunsCount > NUMBER_OF_WORKFLOW_RUNS_TO_KEEP; }, authContext, + { lite: true }, ); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts index 4ad910bc44..91d26a65a0 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts @@ -23,6 +23,8 @@ import { getStaledRunsFindOptions } from 'src/modules/workflow/workflow-runner/w export const WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN = '0 * * * *'; +const WORKSPACE_BATCH_SIZE = 50; + @Processor(MessageQueue.cronQueue) export class WorkflowHandleStaledRunsCronJob { private readonly logger = new Logger(WorkflowHandleStaledRunsCronJob.name); @@ -53,25 +55,30 @@ export class WorkflowHandleStaledRunsCronJob { let enqueuedCount = 0; - for (const workspace of activeWorkspaces) { - try { - const hasStaledRuns = await this.hasStaledRuns(workspace.id); + for ( + let workspaceIndex = 0; + workspaceIndex < activeWorkspaces.length; + workspaceIndex += WORKSPACE_BATCH_SIZE + ) { + const batch = activeWorkspaces.slice( + workspaceIndex, + workspaceIndex + WORKSPACE_BATCH_SIZE, + ); - if (hasStaledRuns) { - await this.messageQueueService.add( - WorkflowHandleStaledRunsJob.name, - { - workspaceId: workspace.id, - }, - ); + const results = await Promise.allSettled( + batch.map((workspace) => this.checkAndEnqueue(workspace.id)), + ); + + for (const [index, result] of results.entries()) { + if (result.status === 'fulfilled' && result.value) { enqueuedCount++; } - } catch (error) { - this.exceptionHandlerService.captureExceptions([error], { - workspace: { - id: workspace.id, - }, - }); + + if (result.status === 'rejected') { + this.exceptionHandlerService.captureExceptions([result.reason], { + workspace: { id: batch[index].id }, + }); + } } } @@ -80,6 +87,21 @@ export class WorkflowHandleStaledRunsCronJob { ); } + private async checkAndEnqueue(workspaceId: string): Promise { + const hasStaledRuns = await this.hasStaledRuns(workspaceId); + + if (hasStaledRuns) { + await this.messageQueueService.add( + WorkflowHandleStaledRunsJob.name, + { workspaceId }, + ); + + return true; + } + + return false; + } + private async hasStaledRuns(workspaceId: string): Promise { const authContext = buildSystemAuthContext(workspaceId); @@ -97,6 +119,7 @@ export class WorkflowHandleStaledRunsCronJob { }); }, authContext, + { lite: true }, ); } } 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 83c2bbb8a8..1f8bf00ae1 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 @@ -23,6 +23,8 @@ import { export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '*/5 * * * *'; +const WORKSPACE_BATCH_SIZE = 10; + @Processor(MessageQueue.cronQueue) export class WorkflowRunEnqueueCronJob { private readonly logger = new Logger(WorkflowRunEnqueueCronJob.name); @@ -53,26 +55,30 @@ export class WorkflowRunEnqueueCronJob { let enqueuedCount = 0; - for (const workspace of activeWorkspaces) { - try { - const hasNotStartedRuns = await this.hasNotStartedRuns(workspace.id); + for ( + let workspaceIndex = 0; + workspaceIndex < activeWorkspaces.length; + workspaceIndex += WORKSPACE_BATCH_SIZE + ) { + const batch = activeWorkspaces.slice( + workspaceIndex, + workspaceIndex + WORKSPACE_BATCH_SIZE, + ); - if (hasNotStartedRuns) { - await this.messageQueueService.add( - WorkflowRunEnqueueJob.name, - { - workspaceId: workspace.id, - isCacheMode: false, - }, - ); + const results = await Promise.allSettled( + batch.map((workspace) => this.checkAndEnqueue(workspace.id)), + ); + + for (const [index, result] of results.entries()) { + if (result.status === 'fulfilled' && result.value) { enqueuedCount++; } - } catch (error) { - this.exceptionHandlerService.captureExceptions([error], { - workspace: { - id: workspace.id, - }, - }); + + if (result.status === 'rejected') { + this.exceptionHandlerService.captureExceptions([result.reason], { + workspace: { id: batch[index].id }, + }); + } } } @@ -81,6 +87,21 @@ export class WorkflowRunEnqueueCronJob { ); } + private async checkAndEnqueue(workspaceId: string): Promise { + const hasNotStartedRuns = await this.hasNotStartedRuns(workspaceId); + + if (hasNotStartedRuns) { + await this.messageQueueService.add( + WorkflowRunEnqueueJob.name, + { workspaceId, isCacheMode: false }, + ); + + return true; + } + + return false; + } + private async hasNotStartedRuns(workspaceId: string): Promise { const authContext = buildSystemAuthContext(workspaceId); @@ -98,6 +119,7 @@ export class WorkflowRunEnqueueCronJob { }); }, authContext, + { lite: true }, ); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts index 5c05bc7f9b..ebf520f01d 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts @@ -9,10 +9,7 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu 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 { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; -import { - WorkflowRunStatus, - WorkflowRunWorkspaceEntity, -} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { NUMBER_OF_WORKFLOW_RUNS_TO_KEEP } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/number-of-workflow-runs-to-keep'; import { RUNS_TO_CLEAN_THRESHOLD_DAYS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/runs-to-clean-threshold'; @@ -37,39 +34,116 @@ export class WorkflowCleanWorkflowRunsJob { const schemaName = getWorkspaceSchemaName(workspaceId); const authContext = buildSystemAuthContext(workspaceId); + this.logger.log( + `[WorkflowCleanWorkflowRunsJob] Starting job for workspace ${workspaceId}`, + ); + await this.globalWorkspaceOrmManager.executeInWorkspaceContext(async () => { - const workflowRunsToDelete = await this.dataSource.query( + const BATCH_SIZE = 200; + let totalDeleted = 0; + + const oldRunsDeleted = await this.deleteOldRuns({ + schemaName, + batchSize: BATCH_SIZE, + }); + + totalDeleted += oldRunsDeleted; + + const excessRunsDeleted = await this.deleteExcessRunsPerWorkflow({ + schemaName, + batchSize: BATCH_SIZE, + }); + + totalDeleted += excessRunsDeleted; + + this.logger.log( + `[WorkflowCleanWorkflowRunsJob] Deleted ${totalDeleted} workflow runs for workspace ${workspaceId}`, + ); + }, authContext); + } + + private async deleteOldRuns({ + schemaName, + batchSize, + }: { + schemaName: string; + batchSize: number; + }): Promise { + let totalDeleted = 0; + let deletedCount: number; + + do { + const result = await this.dataSource.query( + ` + DELETE FROM ${schemaName}."workflowRun" + WHERE id IN ( + SELECT id FROM ${schemaName}."workflowRun" + WHERE status IN ($1, $2) + AND "createdAt" < NOW() - MAKE_INTERVAL(days => $3) + LIMIT $4 + ) + RETURNING id; + `, + [ + WorkflowRunStatus.COMPLETED, + WorkflowRunStatus.FAILED, + RUNS_TO_CLEAN_THRESHOLD_DAYS, + batchSize, + ], + ); + + // TypeORM's dataSource.query() for for DELETE ... RETURNING returns a tuple [rows, affectedCount] + deletedCount = result[0].length; + totalDeleted += deletedCount; + } while (deletedCount > 0); + + return totalDeleted; + } + + private async deleteExcessRunsPerWorkflow({ + schemaName, + batchSize, + }: { + schemaName: string; + batchSize: number; + }): Promise { + let totalDeleted = 0; + let deletedCount: number; + + do { + const result = await this.dataSource.query( ` WITH ranked_runs AS ( SELECT id, ROW_NUMBER() OVER ( PARTITION BY "workflowId" ORDER BY "createdAt" DESC - ) AS rn, - "createdAt" + ) AS rn FROM ${schemaName}."workflowRun" - WHERE status IN ('${WorkflowRunStatus.COMPLETED}', '${WorkflowRunStatus.FAILED}') + WHERE status IN ($1, $2) + ), + runs_to_delete AS ( + SELECT id FROM ranked_runs + WHERE rn > $3 + LIMIT $4 ) - SELECT id, rn FROM ranked_runs - WHERE rn > ${NUMBER_OF_WORKFLOW_RUNS_TO_KEEP} - OR "createdAt" < NOW() - INTERVAL '${RUNS_TO_CLEAN_THRESHOLD_DAYS} days'; + DELETE FROM ${schemaName}."workflowRun" + WHERE id IN (SELECT id FROM runs_to_delete) + RETURNING id; `, + [ + WorkflowRunStatus.COMPLETED, + WorkflowRunStatus.FAILED, + NUMBER_OF_WORKFLOW_RUNS_TO_KEEP, + batchSize, + ], ); - const workflowRunRepository = - await this.globalWorkspaceOrmManager.getRepository( - workspaceId, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); + // TypeORM's dataSource.query() for for DELETE ... RETURNING returns a tuple [rows, affectedCount] + deletedCount = result[0].length; + totalDeleted += deletedCount; + } while (deletedCount > 0); - for (const workflowRunToDelete of workflowRunsToDelete) { - await workflowRunRepository.delete(workflowRunToDelete.id); - } - - this.logger.log( - `Deleted ${workflowRunsToDelete.length} workflow runs for workspace ${workspaceId}`, - ); - }, authContext); + return totalDeleted; } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts index 8ce7fbe623..5d2b4a58a8 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts @@ -277,13 +277,21 @@ export class WorkflowRunWorkspaceService { workspaceId, }); + const existingStepInfos = workflowRunToUpdate.state?.stepInfos ?? {}; + + const mergedStepInfos = { ...existingStepInfos }; + + for (const [stepId, info] of Object.entries(stepInfos)) { + mergedStepInfos[stepId] = { + ...(existingStepInfos[stepId] || {}), + ...info, + }; + } + const partialUpdate = { state: { ...workflowRunToUpdate.state, - stepInfos: { - ...workflowRunToUpdate.state?.stepInfos, - ...stepInfos, - }, + stepInfos: mergedStepInfos, }, };