From 748e614f6c1c425d091d64647913cf5b14c37d54 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Thu, 12 Feb 2026 14:27:36 +0100 Subject: [PATCH] Workflow bug fixes (#17886) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes https://github.com/twentyhq/private-issues/issues/418 Currency filtering was not properly managed. Since this is a select, we were doing 'USD' == [USD]. Replacing by contains as for other select Capture d’écran 2026-02-12 à 11 08 42 Fixes https://github.com/twentyhq/twenty/issues/17611 If-else branches not executed has to be marked as skipped. Otherwise, the iterator will never start the next iteration. It will wait for some not started nodes. Capture d’écran 2026-02-12 à 10 56 45 --- .../utils/getEdgePathStrategy.ts | 3 +- .../utils/evaluate-filter-conditions.util.ts | 4 +- .../get-all-step-ids-in-loop.util.spec.ts | 102 ++++++++++++ .../utils/get-all-step-ids-in-loop.util.ts | 22 ++- ...orkflow-executor.workspace-service.spec.ts | 147 ++++++++++++++++++ .../workflow-executor.workspace-service.ts | 85 ++++++++-- .../workflow-runner/jobs/run-workflow.job.ts | 2 +- .../workflow-clean-workflow-runs.cron.job.ts | 26 +++- .../workflow-handle-staled-runs.cron.job.ts | 26 +++- .../jobs/workflow-run-enqueue.cron.job.ts | 28 ++-- 10 files changed, 405 insertions(+), 40 deletions(-) diff --git a/packages/twenty-front/src/modules/workflow/workflow-diagram/utils/getEdgePathStrategy.ts b/packages/twenty-front/src/modules/workflow/workflow-diagram/utils/getEdgePathStrategy.ts index 71bb0b3967..7c2cb2207b 100644 --- a/packages/twenty-front/src/modules/workflow/workflow-diagram/utils/getEdgePathStrategy.ts +++ b/packages/twenty-front/src/modules/workflow/workflow-diagram/utils/getEdgePathStrategy.ts @@ -12,8 +12,9 @@ export const getEdgePathStrategy = ({ steps: WorkflowStep[]; }) => { const nextStep = steps.find((s) => s.id === nextStepId); + if (!isDefined(nextStep)) { - throw new Error('Expected to find step defined in nextStepIds'); + return undefined; } const useLoopBackToIteratorStyle = diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-filter-conditions.util.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-filter-conditions.util.ts index a4bd084a59..1ba0308573 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-filter-conditions.util.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/filter/utils/evaluate-filter-conditions.util.ts @@ -307,9 +307,9 @@ function evaluateCurrencyFilter(filter: ResolvedFilter): boolean { if (filter.compositeFieldSubFieldName === 'currencyCode') { switch (filter.operand) { case ViewFilterOperand.IS: - return filter.leftOperand === filter.rightOperand; + return contains(filter.leftOperand, filter.rightOperand); case ViewFilterOperand.IS_NOT: - return filter.leftOperand !== filter.rightOperand; + return !contains(filter.leftOperand, filter.rightOperand); case ViewFilterOperand.IS_EMPTY: return !isNonEmptyString(filter.leftOperand); case ViewFilterOperand.IS_NOT_EMPTY: diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/__tests__/get-all-step-ids-in-loop.util.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/__tests__/get-all-step-ids-in-loop.util.spec.ts index 27dd4d7214..21833ea699 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/__tests__/get-all-step-ids-in-loop.util.spec.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/__tests__/get-all-step-ids-in-loop.util.spec.ts @@ -1,10 +1,14 @@ +import { type StepIfElseBranch } from 'twenty-shared/workflow'; + import { type WorkflowCodeActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/code/types/workflow-code-action-settings.type'; +import { type WorkflowIfElseActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/if-else/types/workflow-if-else-action-settings.type'; import { type WorkflowIteratorActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/types/workflow-iterator-action-settings.type'; import { getAllStepIdsInLoop } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/get-all-step-ids-in-loop.util'; import { type WorkflowAction, WorkflowActionType, type WorkflowCodeAction, + type WorkflowIfElseAction, type WorkflowIteratorAction, } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type'; @@ -50,6 +54,30 @@ describe('getAllStepIdsInLoop', () => { }, }); + const createIfElseMockStep = ( + id: string, + branches: StepIfElseBranch[], + nextStepIds: string[] = [], + ): WorkflowIfElseAction => ({ + id, + name: `Step ${id}`, + type: WorkflowActionType.IF_ELSE, + valid: true, + nextStepIds, + settings: { + input: { + stepFilterGroups: [], + stepFilters: [], + branches, + }, + outputSchema: {}, + errorHandlingOptions: { + continueOnFailure: { value: false }, + retryOnFailure: { value: false }, + }, + } as WorkflowIfElseActionSettings, + }); + describe('simple loop scenarios', () => { it('should return all step IDs in a simple linear loop', () => { const steps: WorkflowAction[] = [ @@ -177,6 +205,80 @@ describe('getAllStepIdsInLoop', () => { }); }); + describe('if-else scenarios', () => { + it('should include steps in all if-else branches within a loop', () => { + const steps: WorkflowAction[] = [ + createIteratorMockStep('iterator1', [], ['ifElse1']), + createIfElseMockStep('ifElse1', [ + { id: 'branch-if', filterGroupId: 'fg1', nextStepIds: ['stepA'] }, + { id: 'branch-else', nextStepIds: ['stepB'] }, + ]), + createCodeMockStep('stepA', ['iterator1']), + createCodeMockStep('stepB', ['iterator1']), + ]; + + const result = getAllStepIdsInLoop({ + iteratorStepId: 'iterator1', + initialLoopStepIds: ['ifElse1'], + steps, + }); + + expect(result).toEqual( + expect.arrayContaining(['ifElse1', 'stepA', 'stepB']), + ); + expect(result).toHaveLength(3); + }); + + it('should include deeply nested steps inside if-else branches', () => { + const steps: WorkflowAction[] = [ + createIteratorMockStep('iterator1', [], ['ifElse1']), + createIfElseMockStep('ifElse1', [ + { id: 'branch-if', filterGroupId: 'fg1', nextStepIds: ['stepA'] }, + { id: 'branch-else', nextStepIds: ['stepB'] }, + ]), + createCodeMockStep('stepA', ['stepC']), + createCodeMockStep('stepB', ['stepC']), + createCodeMockStep('stepC', ['iterator1']), + ]; + + const result = getAllStepIdsInLoop({ + iteratorStepId: 'iterator1', + initialLoopStepIds: ['ifElse1'], + steps, + }); + + expect(result).toEqual( + expect.arrayContaining(['ifElse1', 'stepA', 'stepB', 'stepC']), + ); + expect(result).toHaveLength(4); + }); + + it('should handle if-else with steps before and after within a loop', () => { + const steps: WorkflowAction[] = [ + createIteratorMockStep('iterator1', [], ['step1']), + createCodeMockStep('step1', ['ifElse1']), + createIfElseMockStep('ifElse1', [ + { id: 'branch-if', filterGroupId: 'fg1', nextStepIds: ['stepA'] }, + { id: 'branch-else', nextStepIds: ['stepB'] }, + ]), + createCodeMockStep('stepA', ['step2']), + createCodeMockStep('stepB', ['step2']), + createCodeMockStep('step2', ['iterator1']), + ]; + + const result = getAllStepIdsInLoop({ + iteratorStepId: 'iterator1', + initialLoopStepIds: ['step1'], + steps, + }); + + expect(result).toEqual( + expect.arrayContaining(['step1', 'ifElse1', 'stepA', 'stepB', 'step2']), + ); + expect(result).toHaveLength(5); + }); + }); + describe('edge cases', () => { it('should handle empty initial loop step IDs', () => { const steps: WorkflowAction[] = [ diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/get-all-step-ids-in-loop.util.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/get-all-step-ids-in-loop.util.ts index 54c0c796bd..600c53b03a 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/get-all-step-ids-in-loop.util.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/iterator/utils/get-all-step-ids-in-loop.util.ts @@ -1,3 +1,4 @@ +import { isWorkflowIfElseAction } from 'src/modules/workflow/workflow-executor/workflow-actions/if-else/guards/is-workflow-if-else-action.guard'; import { isWorkflowIteratorAction } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/guards/is-workflow-iterator-action.guard'; import { type WorkflowIteratorActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/iterator/types/workflow-iterator-action-settings.type'; import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type'; @@ -25,7 +26,7 @@ const traverseSteps = ({ const step = steps.find((s) => s.id === stepId); - if (!step || !step.nextStepIds) { + if (!step) { continue; } @@ -46,10 +47,27 @@ const traverseSteps = ({ } } + if (isWorkflowIfElseAction(step)) { + for (const branch of step.settings.input.branches) { + if (branch.nextStepIds) { + traverseSteps({ + iteratorStepId, + stepIds: branch.nextStepIds, + steps, + visitedStepIds, + allStepIdsInLoop, + }); + } + } + } + + if (!step.nextStepIds) { + continue; + } + const connectsBackToIterator = step.nextStepIds.includes(iteratorStepId); if (connectsBackToIterator) { - // We've found the end of the loop, stop traversing continue; } 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 3c8935fbc5..5877d619d0 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 @@ -382,6 +382,153 @@ describe('WorkflowExecutorWorkspaceService', () => { }); }); + describe('getNextStepIdsToExecute', () => { + it('should return nextStepIds for a regular step', async () => { + const step = { + id: 'step-1', + type: WorkflowActionType.CODE, + nextStepIds: ['step-2', 'step-3'], + settings: {}, + } as WorkflowAction; + + const result = await service.getNextStepIdsToExecute({ + executedStep: step, + executedStepOutput: { result: {} }, + }); + + expect(result).toEqual({ + nextStepIdsToExecute: ['step-2', 'step-3'], + }); + }); + + it('should return initialLoopStepIds for an iterator that has not processed all items', async () => { + const step = { + id: 'iterator-1', + type: WorkflowActionType.ITERATOR, + nextStepIds: ['after-loop'], + settings: { + input: { + initialLoopStepIds: ['loop-step-1'], + }, + }, + } as WorkflowAction; + + const result = await service.getNextStepIdsToExecute({ + executedStep: step, + executedStepOutput: { + result: { hasProcessedAllItems: false }, + }, + }); + + expect(result).toEqual({ + nextStepIdsToExecute: ['loop-step-1'], + }); + }); + + it('should return nextStepIds for an iterator that has processed all items', async () => { + const step = { + id: 'iterator-1', + type: WorkflowActionType.ITERATOR, + nextStepIds: ['after-loop'], + settings: { + input: { + initialLoopStepIds: ['loop-step-1'], + }, + }, + } as WorkflowAction; + + const result = await service.getNextStepIdsToExecute({ + executedStep: step, + executedStepOutput: { + result: { hasProcessedAllItems: true }, + }, + }); + + expect(result).toEqual({ + nextStepIdsToExecute: ['after-loop'], + }); + }); + + it('should return matching branch nextStepIds and non-matching branch nextStepIds to skip for if-else', async () => { + const step = { + id: 'if-else-1', + type: WorkflowActionType.IF_ELSE, + nextStepIds: [], + settings: { + input: { + branches: [ + { + id: 'branch-if', + filterGroupId: 'fg1', + nextStepIds: ['step-a'], + }, + { + id: 'branch-else', + nextStepIds: ['step-b'], + }, + ], + stepFilterGroups: [], + stepFilters: [], + }, + }, + } as unknown as WorkflowAction; + + const result = await service.getNextStepIdsToExecute({ + executedStep: step, + executedStepOutput: { + result: { matchingBranchId: 'branch-if' }, + }, + }); + + expect(result).toEqual({ + nextStepIdsToExecute: ['step-a'], + nextStepIdsToSkip: ['step-b'], + }); + }); + + it('should skip multiple non-matching branches for if-else with many branches', async () => { + const step = { + id: 'if-else-1', + type: WorkflowActionType.IF_ELSE, + nextStepIds: [], + settings: { + input: { + branches: [ + { + id: 'branch-1', + filterGroupId: 'fg1', + nextStepIds: ['step-a'], + }, + { + id: 'branch-2', + filterGroupId: 'fg2', + nextStepIds: ['step-b'], + }, + { + id: 'branch-else', + nextStepIds: ['step-c'], + }, + ], + stepFilterGroups: [], + stepFilters: [], + }, + }, + } as unknown as WorkflowAction; + + const result = await service.getNextStepIdsToExecute({ + executedStep: step, + executedStepOutput: { + result: { matchingBranchId: 'branch-2' }, + }, + }); + + expect(result).toEqual({ + nextStepIdsToExecute: ['step-b'], + nextStepIdsToSkip: ['step-a', 'step-c'], + }); + }); + }); + describe('sendWorkflowNodeRunEvent', () => { it('should emit a billing event', () => { service['sendWorkflowNodeRunEvent']('workspace-id', 'workflow-id'); 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 d3066e6eb7..ddb6660c51 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 @@ -169,10 +169,21 @@ export class WorkflowExecutorWorkspaceService { return; } - const nextStepIdsToExecute = await this.getNextStepIdsToExecute({ - executedStep: stepToExecute, - executedStepOutput: actionOutput, - }); + const { nextStepIdsToExecute, nextStepIdsToSkip } = + await this.getNextStepIdsToExecute({ + executedStep: stepToExecute, + executedStepOutput: actionOutput, + }); + + if (isDefined(nextStepIdsToSkip) && nextStepIdsToSkip.length > 0) { + await this.skipStepsAndContinue({ + stepIdsToSkip: nextStepIdsToSkip, + steps, + workflowRunId, + workspaceId, + executedStepsCount: (executedStepsCount ?? 0) + 1, + }); + } if (isDefined(nextStepIdsToExecute) && nextStepIdsToExecute.length > 0) { await this.executeFromSteps({ @@ -191,7 +202,10 @@ export class WorkflowExecutorWorkspaceService { }: { executedStep: WorkflowAction; executedStepOutput: WorkflowActionOutput; - }): Promise { + }): Promise<{ + nextStepIdsToExecute?: string[]; + nextStepIdsToSkip?: string[]; + }> { const isIteratorStep = isWorkflowIteratorAction(executedStep); if (isIteratorStep) { @@ -200,9 +214,13 @@ export class WorkflowExecutorWorkspaceService { | undefined; if (!iteratorStepResult?.hasProcessedAllItems) { - return isString(executedStep.settings.input.initialLoopStepIds) + const nextStepIdsToExecute = isString( + executedStep.settings.input.initialLoopStepIds, + ) ? JSON.parse(executedStep.settings.input.initialLoopStepIds) : executedStep.settings.input.initialLoopStepIds; + + return { nextStepIdsToExecute }; } } @@ -212,15 +230,26 @@ export class WorkflowExecutorWorkspaceService { | undefined; if (ifElseResult?.matchingBranchId) { - const matchingBranch = executedStep.settings.input.branches.find( + const branches = executedStep.settings.input.branches; + + const matchingBranch = branches.find( (branch) => branch.id === ifElseResult.matchingBranchId, ); - return matchingBranch?.nextStepIds; + const nonMatchingBranches = branches.filter( + (branch) => branch.id !== ifElseResult.matchingBranchId, + ); + + return { + nextStepIdsToExecute: matchingBranch?.nextStepIds, + nextStepIdsToSkip: nonMatchingBranches.flatMap( + (branch) => branch.nextStepIds, + ), + }; } } - return executedStep.nextStepIds; + return { nextStepIdsToExecute: executedStep.nextStepIds }; } private async computeWorkflowRunStatus({ @@ -415,6 +444,44 @@ export class WorkflowExecutorWorkspaceService { } } + private async skipStepsAndContinue({ + stepIdsToSkip, + steps, + workflowRunId, + workspaceId, + executedStepsCount, + }: { + stepIdsToSkip: string[]; + steps: WorkflowAction[]; + workflowRunId: string; + workspaceId: string; + executedStepsCount: number; + }) { + await Promise.all( + stepIdsToSkip.map(async (stepId) => { + await this.workflowRunWorkspaceService.updateWorkflowRunStepInfo({ + stepId, + stepInfo: { status: StepStatus.SKIPPED }, + workflowRunId, + workspaceId, + }); + + const skippedStep = steps.find((step) => step.id === stepId); + const skippedStepNextStepIds = skippedStep?.nextStepIds ?? []; + + if (skippedStepNextStepIds.length > 0) { + await this.executeFromSteps({ + stepIds: skippedStepNextStepIds, + workflowRunId, + workspaceId, + shouldComputeWorkflowRunStatus: false, + executedStepsCount, + }); + } + }), + ); + } + private async continueExecutionFromStepInAnotherJob({ lastExecutedStepId, workflowRunId, 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 1819de6fb4..0c249542d8 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,7 +156,7 @@ export class RunWorkflowJob { const lastExecutedStepOutput = workflowRun.state?.stepInfos[lastExecutedStepId]; - const nextStepIdsToExecute = + const { nextStepIdsToExecute } = await this.workflowExecutorWorkspaceService.getNextStepIdsToExecute({ executedStep: lastExecutedStep, executedStepOutput: lastExecutedStepOutput, 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 b8977c4914..c37b34614d 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 @@ -5,6 +5,7 @@ import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; import { In, Repository } from 'typeorm'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; @@ -36,6 +37,7 @@ export class WorkflowCleanWorkflowRunsCronJob { @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, + private readonly exceptionHandlerService: ExceptionHandlerService, ) {} @Process(WorkflowCleanWorkflowRunsCronJob.name) @@ -56,16 +58,24 @@ export class WorkflowCleanWorkflowRunsCronJob { let enqueuedCount = 0; for (const workspace of activeWorkspaces) { - const hasRunsToClean = await this.hasRunsToClean(workspace.id); + try { + const hasRunsToClean = await this.hasRunsToClean(workspace.id); - if (hasRunsToClean) { - await this.messageQueueService.add( - WorkflowCleanWorkflowRunsJob.name, - { - workspaceId: workspace.id, + if (hasRunsToClean) { + await this.messageQueueService.add( + WorkflowCleanWorkflowRunsJob.name, + { + workspaceId: workspace.id, + }, + ); + enqueuedCount++; + } + } catch (error) { + this.exceptionHandlerService.captureExceptions([error], { + workspace: { + id: workspace.id, }, - ); - enqueuedCount++; + }); } } 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 ef604ea8bd..4ad910bc44 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 @@ -5,6 +5,7 @@ import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; import { Repository } from 'typeorm'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; @@ -32,6 +33,7 @@ export class WorkflowHandleStaledRunsCronJob { @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, + private readonly exceptionHandlerService: ExceptionHandlerService, ) {} @Process(WorkflowHandleStaledRunsCronJob.name) @@ -52,16 +54,24 @@ export class WorkflowHandleStaledRunsCronJob { let enqueuedCount = 0; for (const workspace of activeWorkspaces) { - const hasStaledRuns = await this.hasStaledRuns(workspace.id); + try { + const hasStaledRuns = await this.hasStaledRuns(workspace.id); - if (hasStaledRuns) { - await this.messageQueueService.add( - WorkflowHandleStaledRunsJob.name, - { - workspaceId: workspace.id, + if (hasStaledRuns) { + await this.messageQueueService.add( + WorkflowHandleStaledRunsJob.name, + { + workspaceId: workspace.id, + }, + ); + enqueuedCount++; + } + } catch (error) { + this.exceptionHandlerService.captureExceptions([error], { + workspace: { + id: workspace.id, }, - ); - enqueuedCount++; + }); } } 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 e0521b3e33..83c2bbb8a8 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 @@ -5,6 +5,7 @@ import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; import { Repository } from 'typeorm'; import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; @@ -32,6 +33,7 @@ export class WorkflowRunEnqueueCronJob { @InjectMessageQueue(MessageQueue.workflowQueue) private readonly messageQueueService: MessageQueueService, private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, + private readonly exceptionHandlerService: ExceptionHandlerService, ) {} @Process(WorkflowRunEnqueueCronJob.name) @@ -52,17 +54,25 @@ export class WorkflowRunEnqueueCronJob { let enqueuedCount = 0; for (const workspace of activeWorkspaces) { - const hasNotStartedRuns = await this.hasNotStartedRuns(workspace.id); + try { + const hasNotStartedRuns = await this.hasNotStartedRuns(workspace.id); - if (hasNotStartedRuns) { - await this.messageQueueService.add( - WorkflowRunEnqueueJob.name, - { - workspaceId: workspace.id, - isCacheMode: false, + if (hasNotStartedRuns) { + await this.messageQueueService.add( + WorkflowRunEnqueueJob.name, + { + workspaceId: workspace.id, + isCacheMode: false, + }, + ); + enqueuedCount++; + } + } catch (error) { + this.exceptionHandlerService.captureExceptions([error], { + workspace: { + id: workspace.id, }, - ); - enqueuedCount++; + }); } }