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

Fix workflow run output empty (#9616)

Browse Source 
- catch error on action execution. We will log the error and return it
in the step
- catch error on workflow run
- remove the catch in the action. All actions should simply throw and
let the executor do the job

<img width="1512" alt="Capture d’écran 2025-01-14 à 17 35 53"
src="https://github.com/user-attachments/assets/dcf79567-a309-45f1-a640-c50b7ac4769b"
/>
This commit is contained in:
Thomas Trompette
2025-01-14 18:18:40 +01:00
committed by GitHub
parent 3e8fee7ff7
commit 87be542185
4 changed files with 93 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts
View File

@ -1,4 +1,4 @@
import { Scope } from '@nestjs/common';
import { Logger, Scope } from '@nestjs/common';
import { EnvironmentService } from 'src/engine/core-modules/environment/environment.service';
import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
@ -23,6 +23,7 @@ export type RunWorkflowJobData = {
@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST })
export class RunWorkflowJob {
private readonly logger = new Logger(RunWorkflowJob.name);
constructor(
private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService,
private readonly workflowExecutorWorkspaceService: WorkflowExecutorWorkspaceService,
@ -39,41 +40,33 @@ export class RunWorkflowJob {
}: RunWorkflowJobData): Promise<void> {
await this.workflowRunWorkspaceService.startWorkflowRun(workflowRunId);
const workflowVersion =
await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail(
workflowVersionId,
);
await this.throttleExecution(workflowVersion.workflowId, workflowRunId);
const { steps, status } =
await this.workflowExecutorWorkspaceService.execute({
currentStepIndex: 0,
steps: workflowVersion.steps || [],
context: {
trigger: payload,
},
output: {
steps: {},
status: WorkflowRunStatus.RUNNING,
},
});
await this.workflowRunWorkspaceService.endWorkflowRun(
workflowRunId,
status,
{
steps,
},
);
}
private async throttleExecution(workflowId: string, workflowRunId: string) {
try {
await this.throttlerService.throttle(
`${workflowId}-workflow-execution`,
this.environmentService.get('WORKFLOW_EXEC_THROTTLE_LIMIT'),
this.environmentService.get('WORKFLOW_EXEC_THROTTLE_TTL'),
const workflowVersion =
await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail(
workflowVersionId,
);
await this.throttleExecution(workflowVersion.workflowId);
const { steps, status } =
await this.workflowExecutorWorkspaceService.execute({
currentStepIndex: 0,
steps: workflowVersion.steps || [],
context: {
trigger: payload,
},
output: {
steps: {},
status: WorkflowRunStatus.RUNNING,
},
});
await this.workflowRunWorkspaceService.endWorkflowRun(
workflowRunId,
status,
{
steps,
},
);
} catch (error) {
await this.workflowRunWorkspaceService.endWorkflowRun(
@ -81,9 +74,20 @@ export class RunWorkflowJob {
WorkflowRunStatus.FAILED,
{
steps: {},
error: 'Workflow execution rate limit exceeded',
error: error.message,
},
);
}
}
private async throttleExecution(workflowId: string) {
try {
await this.throttlerService.throttle(
`${workflowId}-workflow-execution`,
this.environmentService.get('WORKFLOW_EXEC_THROTTLE_LIMIT'),
this.environmentService.get('WORKFLOW_EXEC_THROTTLE_TTL'),
);
} catch (error) {
throw new WorkflowRunException(
'Workflow execution rate limit exceeded',
WorkflowRunExceptionCode.WORKFLOW_RUN_LIMIT_REACHED,