houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve workflow perfs (#18376)

Browse Source 
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.
This commit is contained in:
Thomas Trompette
2026-03-04 18:29:12 +01:00
committed by GitHub
parent c53d281960
commit 911a46aa45
11 changed files with 354 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/__tests__/iterator-action.workflow-action.spec.ts
+6 -6
View File
@@ -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);
packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/iterator.workflow-action.ts
+3 -8
View File
@@ -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<string, WorkflowRunStepInfo>;
}) {
let stepInfosToUpdate: Record<string, WorkflowRunStepInfo> = {};
const workflowRunToUpdate =
await this.workflowRunWorkspaceService.getWorkflowRunOrFail({
workflowRunId,
workspaceId,
});
const stepInfos = workflowRunToUpdate.state.stepInfos;
if (!hasProcessedAllItems) {
const subStepsInfos = await this.buildSubStepInfosReset({
iteratorStepId,
packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service.ts
+33 -31
View File
@@ -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<string, WorkflowRunStepInfo> = {};
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<string>();
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({
packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts
+30 -7
View File
@@ -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({
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-clean-workflow-runs.cron.job.ts
+39 -16
View File
@@ -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<WorkflowCleanWorkflowRunsJobData>(
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<boolean> {
const hasRunsToClean = await this.hasRunsToClean(workspaceId);
if (hasRunsToClean) {
await this.messageQueueService.add<WorkflowCleanWorkflowRunsJobData>(
WorkflowCleanWorkflowRunsJob.name,
{ workspaceId },
);
return true;
}
return false;
}
private async hasRunsToClean(workspaceId: string): Promise<boolean> {
const authContext = buildSystemAuthContext(workspaceId);
@@ -113,6 +135,7 @@ export class WorkflowCleanWorkflowRunsCronJob {
return totalCompletedRunsCount > NUMBER_OF_WORKFLOW_RUNS_TO_KEEP;
},
authContext,
{ lite: true },
);
}
}
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts
+39 -16
View File
@@ -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<WorkflowHandleStaledRunsJobData>(
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<boolean> {
const hasStaledRuns = await this.hasStaledRuns(workspaceId);
if (hasStaledRuns) {
await this.messageQueueService.add<WorkflowHandleStaledRunsJobData>(
WorkflowHandleStaledRunsJob.name,
{ workspaceId },
);
return true;
}
return false;
}
private async hasStaledRuns(workspaceId: string): Promise<boolean> {
const authContext = buildSystemAuthContext(workspaceId);
@@ -97,6 +119,7 @@ export class WorkflowHandleStaledRunsCronJob {
});
},
authContext,
{ lite: true },
);
}
}
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts
+39 -17
View File
@@ -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<WorkflowRunEnqueueJobData>(
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<boolean> {
const hasNotStartedRuns = await this.hasNotStartedRuns(workspaceId);
if (hasNotStartedRuns) {
await this.messageQueueService.add<WorkflowRunEnqueueJobData>(
WorkflowRunEnqueueJob.name,
{ workspaceId, isCacheMode: false },
);
return true;
}
return false;
}
private async hasNotStartedRuns(workspaceId: string): Promise<boolean> {
const authContext = buildSystemAuthContext(workspaceId);
@@ -98,6 +119,7 @@ export class WorkflowRunEnqueueCronJob {
});
},
authContext,
{ lite: true },
);
}
}
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-clean-workflow-runs.job.ts
+99 -25
View File
@@ -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<number> {
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<number> {
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;
}
}
packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts
+12 -4
View File
@@ -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,
},
};