diff --git a/packages/twenty-server/src/engine/core-modules/metrics/metrics.service.ts b/packages/twenty-server/src/engine/core-modules/metrics/metrics.service.ts index 3c9add6721..76a6d3bfc3 100644 --- a/packages/twenty-server/src/engine/core-modules/metrics/metrics.service.ts +++ b/packages/twenty-server/src/engine/core-modules/metrics/metrics.service.ts @@ -85,11 +85,13 @@ export class MetricsService { eventId, attributes, shouldStoreInCache = true, + debugLog, }: { key: MetricsKeys; eventId?: string; attributes?: Attributes; shouldStoreInCache?: boolean; + debugLog?: string; }) { const counter = this.getMeter().createCounter(key); @@ -98,6 +100,10 @@ export class MetricsService { if (shouldStoreInCache && eventId) { this.metricsCacheService.updateCounter(key, [eventId]); } + + if (isDefined(debugLog)) { + this.logger.debug(debugLog); + } } async batchIncrementCounter({ diff --git a/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts b/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts index 8a5964deb6..f36c11c4aa 100644 --- a/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts +++ b/packages/twenty-server/src/engine/core-modules/metrics/types/metrics-keys.type.ts @@ -22,6 +22,7 @@ export enum MetricsKeys { WorkflowRunStopped = 'workflow-run/stopped', WorkflowRunThrottled = 'workflow-run/throttled', WorkflowRunFailedToEnqueue = 'workflow-run/failed/to-enqueue', + WorkflowRunSystemError = 'workflow-run/system-error', AIToolExecutionFailed = 'ai-tool-execution/failed', AIToolExecutionSucceeded = 'ai-tool-execution/succeeded', SchemaVersionMismatch = 'schema-version/mismatch', diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts index 2b17874e96..0a01db6e43 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job.ts @@ -62,14 +62,14 @@ export class ResumeDelayedWorkflowJob { if (!step || !isWorkflowDelayAction(step)) { throw new WorkflowRunException( 'Step not found or is not a delay action', - WorkflowRunExceptionCode.INVALID_INPUT, + WorkflowRunExceptionCode.INVALID_OPERATION, ); } if (stepInfo?.status !== StepStatus.PENDING) { throw new WorkflowRunException( 'Step is not pending', - WorkflowRunExceptionCode.INVALID_INPUT, + WorkflowRunExceptionCode.INVALID_OPERATION, ); } @@ -102,6 +102,7 @@ export class ResumeDelayedWorkflowJob { error instanceof Error ? error.message : `Error during delay resume: ${String(error)}`, + isSystemError: true, }); throw error; diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/tool-executor-workflow-action.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/tool-executor-workflow-action.ts index 57864d2ba3..401335c933 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/tool-executor-workflow-action.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/tool-executor-workflow-action.ts @@ -9,6 +9,10 @@ import { HttpTool } from 'src/engine/core-modules/tool/tools/http-tool/http-tool import { SendEmailTool } from 'src/engine/core-modules/tool/tools/email-tool/send-email-tool'; import { type ToolInput } from 'src/engine/core-modules/tool/types/tool-input.type'; import { type Tool } from 'src/engine/core-modules/tool/types/tool.type'; +import { + WorkflowStepExecutorException, + WorkflowStepExecutorExceptionCode, +} from 'src/modules/workflow/workflow-executor/exceptions/workflow-step-executor.exception'; import { type WorkflowActionInput } from 'src/modules/workflow/workflow-executor/types/workflow-action-input'; import { type WorkflowActionOutput } from 'src/modules/workflow/workflow-executor/types/workflow-action-output.type'; import { type WorkflowSendEmailActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/mail-sender/types/workflow-send-email-action-input.type'; @@ -39,13 +43,19 @@ export class ToolExecutorWorkflowAction implements WorkflowAction { const step = steps.find((step) => step.id === currentStepId); if (!step) { - throw new Error('Step not found'); + throw new WorkflowStepExecutorException( + 'Step not found', + WorkflowStepExecutorExceptionCode.STEP_NOT_FOUND, + ); } const tool = this.toolsByActionType.get(step.type); if (!tool) { - throw new Error(`No tool found for workflow action type: ${step.type}`); + throw new WorkflowStepExecutorException( + `No tool found for workflow action type: ${step.type}`, + WorkflowStepExecutorExceptionCode.INVALID_STEP_TYPE, + ); } let toolInput = step.settings.input; diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts index ab6bc242f3..67cec22de7 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-executor.module.ts @@ -2,6 +2,7 @@ import { Module } from '@nestjs/common'; import { BillingModule } from 'src/engine/core-modules/billing/billing.module'; import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; +import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module'; import { ToolModule } from 'src/engine/core-modules/tool/tool.module'; import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module'; import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory'; @@ -36,6 +37,7 @@ import { WorkflowRunModule } from 'src/modules/workflow/workflow-runner/workflow EmptyActionModule, FeatureFlagModule, ToolModule, + MetricsModule, ], providers: [ WorkflowExecutorWorkspaceService, diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts index 0bc1e77c14..d60ade250f 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workspace-services/__tests__/workflow-executor.workspace-service.spec.ts @@ -6,7 +6,9 @@ import { BILLING_FEATURE_USED } from 'src/engine/core-modules/billing/constants/ import { BILLING_WORKFLOW_EXECUTION_ERROR_MESSAGE } from 'src/engine/core-modules/billing/constants/billing-workflow-execution-error-message.constant'; import { BillingMeterEventName } from 'src/engine/core-modules/billing/enums/billing-meter-event-names'; import { BillingService } from 'src/engine/core-modules/billing/services/billing.service'; +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory'; import { shouldExecuteStep } from 'src/modules/workflow/workflow-executor/utils/should-execute-step.util'; @@ -14,7 +16,6 @@ import { type WorkflowAction, WorkflowActionType, } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type'; -import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { WorkflowExecutorWorkspaceService } from 'src/modules/workflow/workflow-executor/workspace-services/workflow-executor.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; @@ -61,6 +62,10 @@ describe('WorkflowExecutorWorkspaceService', () => { captureExceptions: jest.fn(), }; + const mockMetricsService = { + incrementCounter: jest.fn(), + }; + const mockMessageQueueService = { add: jest.fn(), }; @@ -97,6 +102,10 @@ describe('WorkflowExecutorWorkspaceService', () => { provide: `MESSAGE_QUEUE_${MessageQueue.workflowQueue}`, useValue: mockMessageQueueService, }, + { + provide: MetricsService, + useValue: mockMetricsService, + }, ], }).compile(); 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 cafc46064b..2e759b7b5c 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 @@ -19,6 +19,8 @@ import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handl import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; +import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; import { WorkspaceEventEmitter } from 'src/engine/workspace-event-emitter/workspace-event-emitter'; import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { workflowHasRunningSteps } from 'src/modules/workflow/common/utils/workflow-has-running-steps.util'; @@ -57,6 +59,7 @@ export class WorkflowExecutorWorkspaceService { private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService, private readonly billingService: BillingService, private readonly exceptionHandlerService: ExceptionHandlerService, + private readonly metricsService: MetricsService, @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, ) {} @@ -70,7 +73,7 @@ export class WorkflowExecutorWorkspaceService { }: WorkflowExecutorInput) { await Promise.all( stepIds.map(async (stepIdToExecute) => { - await this.executeFromStep({ + return this.executeFromStep({ stepId: stepIdToExecute, workflowRunId, workspaceId, @@ -92,7 +95,7 @@ export class WorkflowExecutorWorkspaceService { workflowRunId, workspaceId, executedStepsCount, - }: WorkflowBranchExecutorInput) { + }: WorkflowBranchExecutorInput): Promise { const workflowRun = await this.workflowRunWorkspaceService.getWorkflowRunOrFail({ workflowRunId, @@ -111,6 +114,7 @@ export class WorkflowExecutorWorkspaceService { workspaceId, status: WorkflowRunStatus.FAILED, error: 'Step not found', + isSystemError: true, }); return; @@ -500,6 +504,12 @@ export class WorkflowExecutorWorkspaceService { this.exceptionHandlerService.captureExceptions([error], { workspace: { id: workspaceId }, }); + + await this.metricsService.incrementCounter({ + key: MetricsKeys.WorkflowRunSystemError, + eventId: workflowRunId, + debugLog: `[Workflow Run System Error] Workflow run ${workflowRunId} in workspace ${workspaceId} ended with system error`, + }); } return { 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 9449065da8..09f5afd463 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 @@ -61,6 +61,7 @@ export class RunWorkflowJob { workflowRunId, status: WorkflowRunStatus.FAILED, error: error.message, + isSystemError: true, }); throw error; 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 5d2b4a58a8..e69cf1cf03 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 @@ -187,11 +187,13 @@ export class WorkflowRunWorkspaceService { workspaceId, status, error, + isSystemError, }: { workflowRunId: string; workspaceId: string; status: Extract; error?: string; + isSystemError?: boolean; }) { const workflowRunToUpdate = await this.getWorkflowRunOrFail({ workflowRunId, @@ -216,15 +218,25 @@ export class WorkflowRunWorkspaceService { await this.updateWorkflowRun({ workflowRunId, workspaceId, partialUpdate }); + const metricKey = + status === WorkflowRunStatus.COMPLETED + ? MetricsKeys.WorkflowRunCompleted + : status === WorkflowRunStatus.STOPPED + ? MetricsKeys.WorkflowRunStopped + : MetricsKeys.WorkflowRunFailed; + await this.metricsService.incrementCounter({ - key: - status === WorkflowRunStatus.COMPLETED - ? MetricsKeys.WorkflowRunCompleted - : status === WorkflowRunStatus.STOPPED - ? MetricsKeys.WorkflowRunStopped - : MetricsKeys.WorkflowRunFailed, + key: metricKey, eventId: workflowRunId, }); + + if (isSystemError) { + await this.metricsService.incrementCounter({ + key: MetricsKeys.WorkflowRunSystemError, + eventId: workflowRunId, + debugLog: `[Workflow Run System Error] Workflow run ${workflowRunId} in workspace ${workspaceId} ended with system error`, + }); + } } @WithLock('workflowRunId') @@ -250,7 +262,7 @@ export class WorkflowRunWorkspaceService { stepInfos: { ...workflowRunToUpdate.state?.stepInfos, [stepId]: { - ...(workflowRunToUpdate.state?.stepInfos[stepId] || {}), + ...workflowRunToUpdate.state?.stepInfos[stepId], result: stepInfo?.result, error: stepInfo?.error, status: stepInfo.status, @@ -283,7 +295,7 @@ export class WorkflowRunWorkspaceService { for (const [stepId, info] of Object.entries(stepInfos)) { mergedStepInfos[stepId] = { - ...(existingStepInfos[stepId] || {}), + ...existingStepInfos[stepId], ...info, }; } @@ -335,7 +347,7 @@ export class WorkflowRunWorkspaceService { state: { ...workflowRunToUpdate.state, flow: { - ...(workflowRunToUpdate.state?.flow ?? {}), + ...workflowRunToUpdate.state?.flow, steps: updatedSteps, }, },