From 98c71d7b3df3fc61b4ca14d5510a05bc3bbde8b8 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Mon, 20 Jul 2026 14:21:33 +0200 Subject: [PATCH] fix(server): recover workflow runs whose queue job was lost - monitoring only (#22995) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Context A workflow run can get permanently stuck in RUNNING when the queue job executing a step dies without failing (worker crash/restart, lost BullMQ job). The step stays `RUNNING` in the persisted state, nothing ever recomputes the run status, and the run never terminates. If a user clicks Stop, it wedges in STOPPING instead (the stuck-STOPPING sweeper from #22900 then catches it after 1h, but only because of the manual stop). Self-hosters also have no way to tell that a job was lost, or when. ## What this PR does ### Detect runs stuck in RUNNING (monitoring only, no finalization yet) New `handleStuckRunningRunsForWorkspace` in the staled-runs sweeper (same cron/job/CLI wiring as the stuck-STOPPING recovery): - Targets RUNNING runs with `updatedAt` older than 1h (`updatedAt` refreshes on every step-info write, so staleness means zero progress). - Skips any run that still has a job in the queue: new `getInFlightJobs` on the message queue driver (active/waiting/waiting-children/paused/prioritized/delayed), matched by run-id-prefixed job id with a `job.data.workflowRunId` fallback for jobs enqueued before this deploys. - A truly orphaned run (orphaned RUNNING step, lost between two steps, failed branch, or finished-but-never-finalized) is **flagged, not finalized**: warn log + `WorkflowRunStuckRunningDetected` metric + entry in a per-workspace cache. - On every subsequent sweep, flagged runs are re-checked. One that ended or got a new queue job on its own is recorded as `WorkflowRunStuckRunningFalsePositive` (warn log with the status it reached) and unflagged. The cron keeps sweeping a workspace as long as it has flagged runs. This validates the detection before it is allowed to act: if flagged runs never resolve on their own (no false positives) while `Detected` counts real incidents, a follow-up PR can turn the flag into an actual finalization (fail with a clear "job lost" error so Retry works). Runs waiting on PENDING steps (delay, form) are never flagged. ### Make queue jobs traceable to their run All RunWorkflowJob dispatches now set the job id prefix to the workflow run id, so BullMQ job ids become `-`. Worker logs (`Processing job ` / `processed`) and Redis job keys are now greppable by run id. A new opt-in `allowDuplicatedPrefixes` queue option bypasses the one-waiting-job-per-id dedup (which would otherwise drop parallel-branch continuations); existing `id` users keep dedup by default. ### Observability - `stalled` worker event listener: warn log + new `JobStalled` metric — emitted when BullMQ detects a job whose worker stopped renewing its lock (i.e. died mid-job). - `WorkflowRunStuckRunningDetected` / `WorkflowRunStuckRunningFalsePositive` metrics as described above. ## Out of scope (follow-ups) - Actually finalizing flagged runs once monitoring shows no false positives. - Recovering lost *delayed* resume jobs (PENDING delay step whose scheduled job vanished). - Persisting job ids on the run entity — unnecessary given derived ids. ## Testing - 11 unit tests for the monitoring sweeper (flagging, id-prefix + data fallback in-flight guards, pending skip, failed-branch precedence, false-positive tracking, still-stuck retention, error isolation, never-finalizes) plus find-options specs; 613 tests pass across workflow and message-queue modules. - Not covered: end-to-end kill-the-worker scenario against a real queue. --- .../message-queue/drivers/bullmq.driver.ts | 47 ++- .../interfaces/job-options.interface.ts | 1 + .../message-queue-driver.interface.ts | 8 + .../services/message-queue.service.ts | 15 +- .../metrics/types/metrics-keys.type.ts | 3 + .../delay/delay.workflow-action.ts | 2 + .../delay/jobs/resume-delayed-workflow.job.ts | 2 + ...orkflow-executor.workspace-service.spec.ts | 1 + .../workflow-executor.workspace-service.ts | 2 + .../build-run-workflow-job-options.util.ts | 8 + .../workflow-handle-staled-runs.command.ts | 4 + .../constants/stuck-running-runs-threshold.ts | 1 + .../workflow-handle-staled-runs.cron.job.ts | 43 ++- .../jobs/workflow-handle-staled-runs.job.ts | 3 + ...uck-running-runs-find-options.util.spec.ts | 14 + ...et-stuck-running-runs-find-options.util.ts | 19 + ...uck-running-runs-monitor-cache-key.util.ts | 2 + ...ndle-staled-runs.workspace-service.spec.ts | 329 ++++++++++++++++++ ...ow-handle-staled-runs.workspace-service.ts | 249 +++++++++++-- .../workflow-run-enqueue.workspace-service.ts | 2 + .../workflow-runner.workspace-service.ts | 4 + 21 files changed, 720 insertions(+), 39 deletions(-) create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-running-runs-threshold.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/__tests__/get-stuck-running-runs-find-options.util.spec.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-find-options.util.ts create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-monitor-cache-key.util.ts diff --git a/packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts b/packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts index dad792fc2e..177e33e042 100644 --- a/packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts +++ b/packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts @@ -19,8 +19,14 @@ import { type QueueCronJobOptions, type QueueJobOptions, } from 'src/engine/core-modules/message-queue/drivers/interfaces/job-options.interface'; -import { type MessageQueueDriver } from 'src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface'; -import { type MessageQueueJob } from 'src/engine/core-modules/message-queue/interfaces/message-queue-job.interface'; +import { + type InFlightQueueJob, + type MessageQueueDriver, +} from 'src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface'; +import { + type MessageQueueJob, + type MessageQueueJobData, +} from 'src/engine/core-modules/message-queue/interfaces/message-queue-job.interface'; import { type MessageQueueWorkerOptions } from 'src/engine/core-modules/message-queue/interfaces/message-queue-worker-options.interface'; import { QUEUE_RETENTION } from 'src/engine/core-modules/message-queue/constants/queue-retention.constants'; @@ -250,6 +256,18 @@ export class BullMQDriver shouldStoreInCache: false, }); }); + + this.workerMap[queueName].on('stalled', (jobId) => { + this.logger.warn( + `Job ${jobId} stalled on queue ${queueName}: its worker stopped processing it without completing or failing it`, + ); + + void this.metricsService.incrementCounterForEvent({ + key: MetricsKeys.JobStalled, + attributes: { queue: queueName }, + shouldStoreInCache: false, + }); + }); } async addCron({ @@ -322,7 +340,7 @@ export class BullMQDriver } // This ensures only one waiting job can be queued for a specific option.id - if (options?.id) { + if (options?.id && !options?.allowDuplicatedPrefixes) { const waitingJobs = await this.queueMap[queueName].getJobs(['waiting']); const isJobAlreadyWaiting = waitingJobs.some( @@ -351,4 +369,27 @@ export class BullMQDriver await this.queueMap[queueName].add(jobName, data, queueOptions); } + + async getInFlightJobs( + queueName: MessageQueue, + ): Promise[]> { + if (!this.queueMap[queueName]) { + throw new Error( + `Queue ${queueName} is not registered, make sure you have added it as a queue provider`, + ); + } + + const jobs = await this.queueMap[queueName].getJobs([ + 'active', + 'waiting', + 'waiting-children', + 'paused', + 'prioritized', + 'delayed', + ]); + + return jobs + .filter(isDefined) + .map((job) => ({ id: job.id, data: job.data })); + } } diff --git a/packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/job-options.interface.ts b/packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/job-options.interface.ts index 9df438081a..ee2f74919f 100644 --- a/packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/job-options.interface.ts +++ b/packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/job-options.interface.ts @@ -1,5 +1,6 @@ export interface QueueJobOptions { id?: string; + allowDuplicatedPrefixes?: boolean; priority?: number; retryLimit?: number; delay?: number; diff --git a/packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface.ts b/packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface.ts index 117c4d7ade..8107f31c68 100644 --- a/packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface.ts +++ b/packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface.ts @@ -42,4 +42,12 @@ export interface MessageQueueDriver { jobId?: string; }): Promise; register?(queueName: MessageQueue): void; + getInFlightJobs?( + queueName: MessageQueue, + ): Promise[]>; +} + +export interface InFlightQueueJob { + id?: string; + data: T; } diff --git a/packages/twenty-server/src/engine/core-modules/message-queue/services/message-queue.service.ts b/packages/twenty-server/src/engine/core-modules/message-queue/services/message-queue.service.ts index 9f13086e33..71bc29801d 100644 --- a/packages/twenty-server/src/engine/core-modules/message-queue/services/message-queue.service.ts +++ b/packages/twenty-server/src/engine/core-modules/message-queue/services/message-queue.service.ts @@ -4,7 +4,10 @@ import { type QueueCronJobOptions, type QueueJobOptions, } from 'src/engine/core-modules/message-queue/drivers/interfaces/job-options.interface'; -import { MessageQueueDriver } from 'src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface'; +import { + type InFlightQueueJob, + MessageQueueDriver, +} from 'src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface'; import { type MessageQueueJobData, type MessageQueueJob, @@ -35,6 +38,16 @@ export class MessageQueueService { return this.driver.add(this.queueName, jobName, data, options); } + getInFlightJobs(): Promise< + InFlightQueueJob[] + > { + if (typeof this.driver.getInFlightJobs !== 'function') { + return Promise.resolve([]); + } + + return this.driver.getInFlightJobs(this.queueName); + } + addCron({ jobName, data, 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 d1a12227dc..8d189bc227 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 @@ -23,6 +23,8 @@ export enum MetricsKeys { WorkflowRunThrottled = 'workflow-run/throttled', WorkflowRunFailedToEnqueue = 'workflow-run/failed/to-enqueue', WorkflowRunSystemError = 'workflow-run/system-error', + WorkflowRunStuckRunningDetected = 'workflow-run/stuck-running/detected', + WorkflowRunStuckRunningFalsePositive = 'workflow-run/stuck-running/false-positive', AiChatToolExecutionSucceeded = 'ai-chat/tool-execution-succeeded', AiChatToolExecutionFailed = 'ai-chat/tool-execution-failed', WorkflowAgentToolExecutionSucceeded = 'workflow-agent/tool-execution-succeeded', @@ -52,6 +54,7 @@ export enum MetricsKeys { CommonApiQueryRateLimited = 'common-api-query/rate-limited', JobCompleted = 'job/completed', JobFailed = 'job/failed', + JobStalled = 'job/stalled', JobWaiting = 'job/waiting', JobLatencyMs = 'job/latency-ms', AiChatTurnLatencyMs = 'ai-chat/turn-latency-ms', diff --git a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/delay.workflow-action.ts b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/delay.workflow-action.ts index df95a02769..85cb7de3ba 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/delay.workflow-action.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/delay/delay.workflow-action.ts @@ -18,6 +18,7 @@ import { RESUME_DELAYED_WORKFLOW_JOB_NAME } from 'src/modules/workflow/workflow- import { isWorkflowDelayAction } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/guards/is-workflow-delay-action.guard'; import { ResumeDelayedWorkflowJobData } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/types/resume-delayed-workflow-job-data.type'; import { WorkflowDelayActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/types/workflow-delay-action-input.type'; +import { buildRunWorkflowJobOptions } from 'src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util'; @Injectable() export class DelayWorkflowAction implements WorkflowAction { @@ -105,6 +106,7 @@ export class DelayWorkflowAction implements WorkflowAction { stepId: currentStepId, }, { + ...buildRunWorkflowJobOptions(runInfo.workflowRunId), delay: delayInMs, }, ); 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 0a01db6e43..79e3d08bd7 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 @@ -19,6 +19,7 @@ import { } from 'src/modules/workflow/workflow-runner/exceptions/workflow-run.exception'; import { RunWorkflowJob } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; +import { buildRunWorkflowJobOptions } from 'src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; @Processor({ @@ -92,6 +93,7 @@ export class ResumeDelayedWorkflowJob { workflowRunId, lastExecutedStepId: stepId, }, + buildRunWorkflowJobOptions(workflowRunId), ); } catch (error) { await this.workflowRunWorkspaceService.endWorkflowRun({ 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 2413e0e00b..628097aa04 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 @@ -429,6 +429,7 @@ describe('WorkflowExecutorWorkspaceService', () => { workflowRunId: mockWorkflowRunId, lastExecutedStepId: 'step-1', }, + { id: mockWorkflowRunId, allowDuplicatedPrefixes: true }, ); // Should not execute the next step (step-2) in the same job 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 bf09898df9..1bce3310f1 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 @@ -49,6 +49,7 @@ import { getNextStepIdsForIterator } from 'src/modules/workflow/workflow-executo import { WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type'; import { RUN_WORKFLOW_JOB_NAME } from 'src/modules/workflow/workflow-runner/constants/run-workflow-job-name'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; +import { buildRunWorkflowJobOptions } from 'src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; const MAX_EXECUTED_STEPS_COUNT = 20; @@ -612,6 +613,7 @@ export class WorkflowExecutorWorkspaceService { workflowRunId, lastExecutedStepId, }, + buildRunWorkflowJobOptions(workflowRunId), ); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util.ts new file mode 100644 index 0000000000..52d20c0a12 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util.ts @@ -0,0 +1,8 @@ +import { type QueueJobOptions } from 'src/engine/core-modules/message-queue/drivers/interfaces/job-options.interface'; + +export const buildRunWorkflowJobOptions = ( + workflowRunId: string, +): QueueJobOptions => ({ + id: workflowRunId, + allowDuplicatedPrefixes: true, +}); diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command.ts index 36e1b0f2c1..5bfff6560c 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command.ts @@ -53,6 +53,10 @@ export class WorkflowHandleStaledRunsCommand extends CommandRunner { await this.workflowHandleStaledRunsWorkspaceService.handleStuckStoppingRunsForWorkspace( workspaceId, ); + + await this.workflowHandleStaledRunsWorkspaceService.handleStuckRunningRunsForWorkspace( + workspaceId, + ); } catch (error) { this.logger.error( `Failed to handle staled runs for workspace ${workspaceId}`, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-running-runs-threshold.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-running-runs-threshold.ts new file mode 100644 index 0000000000..4db27d06c3 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-running-runs-threshold.ts @@ -0,0 +1 @@ +export const STUCK_RUNNING_RUNS_THRESHOLD_MS = 60 * 60 * 1000; // 1 hour 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 366839d5d8..e8aa33329a 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 @@ -1,6 +1,7 @@ import { Logger } from '@nestjs/common'; import { InjectDataSource, InjectRepository } from '@nestjs/typeorm'; +import { isDefined } from 'twenty-shared/utils'; import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; import { DataSource, Repository } from 'typeorm'; @@ -18,7 +19,9 @@ import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.ent import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { STALED_RUNS_THRESHOLD_MS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/staled-runs-threshold'; +import { STUCK_RUNNING_RUNS_THRESHOLD_MS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-running-runs-threshold'; import { STUCK_STOPPING_RUNS_THRESHOLD_MS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-stopping-runs-threshold'; +import { getStuckRunningRunsMonitorCacheKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-monitor-cache-key.util'; import { WorkflowHandleStaledRunsJob, WorkflowHandleStaledRunsJobData, @@ -101,12 +104,24 @@ export class WorkflowHandleStaledRunsCronJob { } private async checkAndEnqueue(workspaceId: string): Promise { - const [hasStaledRuns, hasStuckStoppingRuns] = await Promise.all([ + const [ + hasStaledRuns, + hasStuckStoppingRuns, + hasStuckRunningRuns, + hasFlaggedStuckRunningRuns, + ] = await Promise.all([ this.hasStaledRuns(workspaceId), this.hasStuckStoppingRuns(workspaceId), + this.hasStuckRunningRuns(workspaceId), + this.hasFlaggedStuckRunningRuns(workspaceId), ]); - if (hasStaledRuns || hasStuckStoppingRuns) { + if ( + hasStaledRuns || + hasStuckStoppingRuns || + hasStuckRunningRuns || + hasFlaggedStuckRunningRuns + ) { await this.messageQueueService.add( WorkflowHandleStaledRunsJob.name, { workspaceId }, @@ -158,4 +173,28 @@ export class WorkflowHandleStaledRunsCronJob { return result.length > 0; } + + private async hasStuckRunningRuns(workspaceId: string): Promise { + const schemaName = getWorkspaceSchemaName(workspaceId); + const thresholdDate = new Date( + Date.now() - STUCK_RUNNING_RUNS_THRESHOLD_MS, + ); + + const result = await this.coreDataSource.query( + `SELECT 1 FROM ${schemaName}."workflowRun" WHERE "status" = $1 AND "updatedAt" < $2 LIMIT 1`, + [WorkflowRunStatus.RUNNING, thresholdDate], + ); + + return result.length > 0; + } + + private async hasFlaggedStuckRunningRuns( + workspaceId: string, + ): Promise { + const flaggedRuns = await this.cacheStorageService.get< + Record + >(getStuckRunningRunsMonitorCacheKey(workspaceId)); + + return isDefined(flaggedRuns) && Object.keys(flaggedRuns).length > 0; + } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts index 02aea3dcc3..1ca26fc5c0 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts @@ -26,6 +26,9 @@ export class WorkflowHandleStaledRunsJob { this.workflowHandleStaledRunsWorkspaceService.handleStuckStoppingRunsForWorkspace( workspaceId, ), + this.workflowHandleStaledRunsWorkspaceService.handleStuckRunningRunsForWorkspace( + workspaceId, + ), ]); } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/__tests__/get-stuck-running-runs-find-options.util.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/__tests__/get-stuck-running-runs-find-options.util.spec.ts new file mode 100644 index 0000000000..bb41bee7a6 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/__tests__/get-stuck-running-runs-find-options.util.spec.ts @@ -0,0 +1,14 @@ +import { FindOperator } from 'typeorm'; + +import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { getStuckRunningRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-find-options.util'; + +describe('getStuckRunningRunsFindOptions', () => { + it('should match RUNNING runs older than the threshold', () => { + const where = getStuckRunningRunsFindOptions(); + + expect(where.status).toBe(WorkflowRunStatus.RUNNING); + expect(where.updatedAt).toBeInstanceOf(FindOperator); + expect((where.updatedAt as FindOperator).type).toBe('lessThan'); + }); +}); diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-find-options.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-find-options.util.ts new file mode 100644 index 0000000000..59db9a0c87 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-find-options.util.ts @@ -0,0 +1,19 @@ +import { type FindOptionsWhere, LessThan } from 'typeorm'; + +import { + WorkflowRunStatus, + type WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { STUCK_RUNNING_RUNS_THRESHOLD_MS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-running-runs-threshold'; + +export const getStuckRunningRunsFindOptions = + (): FindOptionsWhere => { + const thresholdDate = new Date( + Date.now() - STUCK_RUNNING_RUNS_THRESHOLD_MS, + ); + + return { + status: WorkflowRunStatus.RUNNING, + updatedAt: LessThan(thresholdDate.toISOString()), + }; + }; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-monitor-cache-key.util.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-monitor-cache-key.util.ts new file mode 100644 index 0000000000..551923734a --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-monitor-cache-key.util.ts @@ -0,0 +1,2 @@ +export const getStuckRunningRunsMonitorCacheKey = (workspaceId: string) => + `workflow-stuck-running-runs-monitor:${workspaceId}`; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/__tests__/workflow-handle-staled-runs.workspace-service.spec.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/__tests__/workflow-handle-staled-runs.workspace-service.spec.ts index 29492e1dd9..5888cae017 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/__tests__/workflow-handle-staled-runs.workspace-service.spec.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/__tests__/workflow-handle-staled-runs.workspace-service.spec.ts @@ -1,5 +1,11 @@ import { Test, type TestingModule } from '@nestjs/testing'; +import { StepStatus } from 'twenty-shared/workflow'; + +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; +import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { WorkflowHandleStaledRunsWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service'; @@ -26,6 +32,20 @@ const mockWorkflowThrottlingWorkspaceService = { const mockWorkflowRunWorkspaceService = { endWorkflowRun: jest.fn().mockResolvedValue(undefined), + getWorkflowRunOrFail: jest.fn(), +}; + +const mockMessageQueueService = { + getInFlightJobs: jest.fn().mockResolvedValue([]), +}; + +const mockMetricsService = { + incrementCounterForEvent: jest.fn().mockResolvedValue(undefined), +}; + +const mockCacheStorageService = { + get: jest.fn().mockResolvedValue(undefined), + set: jest.fn().mockResolvedValue(undefined), }; // Mirrors QUERY_MAX_RECORDS, the per-batch cap the service applies. Kept as a @@ -61,9 +81,29 @@ describe('WorkflowHandleStaledRunsWorkspaceService', () => { provide: WorkflowRunWorkspaceService, useValue: mockWorkflowRunWorkspaceService, }, + { + provide: `MESSAGE_QUEUE_${MessageQueue.workflowQueue}`, + useValue: mockMessageQueueService, + }, + { + provide: MetricsService, + useValue: mockMetricsService, + }, + { + provide: CacheStorageNamespace.ModuleWorkflow, + useValue: mockCacheStorageService, + }, ], }).compile(); + mockWorkflowRunWorkspaceService.endWorkflowRun + .mockReset() + .mockResolvedValue(undefined); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockReset(); + mockMessageQueueService.getInFlightJobs.mockReset().mockResolvedValue([]); + mockCacheStorageService.get.mockReset().mockResolvedValue(undefined); + mockCacheStorageService.set.mockReset().mockResolvedValue(undefined); + service = module.get( WorkflowHandleStaledRunsWorkspaceService, ); @@ -237,4 +277,293 @@ describe('WorkflowHandleStaledRunsWorkspaceService', () => { }); }); }); + + describe('handleStuckRunningRunsForWorkspace', () => { + const buildRunningWorkflowRun = ({ + id = 'run-0', + steps, + stepInfos, + }: { + id?: string; + // oxlint-disable-next-line typescript/no-explicit-any + steps: any[]; + // oxlint-disable-next-line typescript/no-explicit-any + stepInfos: Record; + }) => ({ + id, + status: WorkflowRunStatus.RUNNING, + updatedAt: '2026-07-15T00:00:00.000Z', + state: { flow: { steps }, stepInfos }, + }); + + const expectNoWorkflowRunEnded = () => { + expect( + mockWorkflowRunWorkspaceService.endWorkflowRun, + ).not.toHaveBeenCalled(); + }; + + it('should do nothing when there are no stuck running nor flagged runs', async () => { + mockRepository.find.mockResolvedValueOnce([]); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect(mockMessageQueueService.getInFlightJobs).not.toHaveBeenCalled(); + expect(mockCacheStorageService.set).not.toHaveBeenCalled(); + expectNoWorkflowRunEnded(); + }); + + it('should not flag runs that still have an in-flight job matched by id prefix', async () => { + mockRepository.find.mockResolvedValueOnce([{ id: 'run-0' }]); + mockMessageQueueService.getInFlightJobs.mockResolvedValueOnce([ + { id: 'run-0-8a521c10-92b1-4013-a4a7-71f20b1a4a4a', data: {} }, + ]); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect( + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail, + ).not.toHaveBeenCalled(); + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + {}, + ); + expectNoWorkflowRunEnded(); + }); + + it('should not flag runs whose pre-deploy job is matched by job data', async () => { + mockRepository.find.mockResolvedValueOnce([{ id: 'run-0' }]); + mockMessageQueueService.getInFlightJobs.mockResolvedValueOnce([ + { id: '12345', data: { workspaceId, workflowRunId: 'run-0' } }, + ]); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect( + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail, + ).not.toHaveBeenCalled(); + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + {}, + ); + expectNoWorkflowRunEnded(); + }); + + it('should flag a run with an orphaned running step without ending it', async () => { + mockRepository.find.mockResolvedValueOnce([{ id: 'run-0' }]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + buildRunningWorkflowRun({ + steps: [{ id: 'step-1', type: 'CODE', nextStepIds: [] }], + stepInfos: { 'step-1': { status: StepStatus.RUNNING } }, + }), + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect(mockMetricsService.incrementCounterForEvent).toHaveBeenCalledWith( + expect.objectContaining({ + key: MetricsKeys.WorkflowRunStuckRunningDetected, + }), + ); + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + { 'run-0': expect.any(String) }, + ); + expectNoWorkflowRunEnded(); + }); + + it('should not flag runs waiting on a pending step', async () => { + mockRepository.find.mockResolvedValueOnce([{ id: 'run-0' }]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + buildRunningWorkflowRun({ + steps: [{ id: 'step-1', type: 'DELAY', nextStepIds: [] }], + stepInfos: { 'step-1': { status: StepStatus.PENDING } }, + }), + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect( + mockMetricsService.incrementCounterForEvent, + ).not.toHaveBeenCalled(); + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + {}, + ); + expectNoWorkflowRunEnded(); + }); + + it('should flag a run with a failed branch even when another branch is pending', async () => { + mockRepository.find.mockResolvedValueOnce([{ id: 'run-0' }]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + buildRunningWorkflowRun({ + steps: [ + { id: 'step-1', type: 'CODE', nextStepIds: [] }, + { id: 'step-2', type: 'DELAY', nextStepIds: [] }, + ], + stepInfos: { + 'step-1': { status: StepStatus.FAILED }, + 'step-2': { status: StepStatus.PENDING }, + }, + }), + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect(mockMetricsService.incrementCounterForEvent).toHaveBeenCalledWith( + expect.objectContaining({ + key: MetricsKeys.WorkflowRunStuckRunningDetected, + }), + ); + expectNoWorkflowRunEnded(); + }); + + it('should flag a run whose job was lost between two steps', async () => { + mockRepository.find.mockResolvedValueOnce([{ id: 'run-0' }]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + buildRunningWorkflowRun({ + steps: [ + { id: 'step-1', type: 'CODE', nextStepIds: ['step-2'] }, + { id: 'step-2', type: 'CODE', nextStepIds: [] }, + ], + stepInfos: { + 'step-1': { status: StepStatus.SUCCESS }, + 'step-2': { status: StepStatus.NOT_STARTED }, + }, + }), + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect(mockMetricsService.incrementCounterForEvent).toHaveBeenCalledWith( + expect.objectContaining({ + key: MetricsKeys.WorkflowRunStuckRunningDetected, + }), + ); + expectNoWorkflowRunEnded(); + }); + + it('should not flag runs that progressed since the query', async () => { + mockRepository.find.mockResolvedValueOnce([{ id: 'run-0' }]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + { + id: 'run-0', + status: WorkflowRunStatus.COMPLETED, + }, + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect( + mockMetricsService.incrementCounterForEvent, + ).not.toHaveBeenCalled(); + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + {}, + ); + expectNoWorkflowRunEnded(); + }); + + it('should keep a still-stuck flagged run without counting it again', async () => { + mockCacheStorageService.get.mockResolvedValueOnce({ + 'run-0': '2026-07-16T00:00:00.000Z', + }); + mockRepository.find.mockResolvedValueOnce([{ id: 'run-0' }]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + buildRunningWorkflowRun({ + steps: [{ id: 'step-1', type: 'CODE', nextStepIds: [] }], + stepInfos: { 'step-1': { status: StepStatus.RUNNING } }, + }), + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect( + mockMetricsService.incrementCounterForEvent, + ).not.toHaveBeenCalled(); + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + { 'run-0': '2026-07-16T00:00:00.000Z' }, + ); + expectNoWorkflowRunEnded(); + }); + + it('should record a false positive when a flagged run ended on its own', async () => { + mockCacheStorageService.get.mockResolvedValueOnce({ + 'run-0': '2026-07-16T00:00:00.000Z', + }); + mockRepository.find.mockResolvedValueOnce([]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + { + id: 'run-0', + status: WorkflowRunStatus.COMPLETED, + }, + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect(mockMetricsService.incrementCounterForEvent).toHaveBeenCalledWith( + expect.objectContaining({ + key: MetricsKeys.WorkflowRunStuckRunningFalsePositive, + }), + ); + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + {}, + ); + expectNoWorkflowRunEnded(); + }); + + it('should record a false positive when a flagged run got a new queue job', async () => { + mockCacheStorageService.get.mockResolvedValueOnce({ + 'run-0': '2026-07-16T00:00:00.000Z', + }); + mockRepository.find.mockResolvedValueOnce([]); + mockMessageQueueService.getInFlightJobs.mockResolvedValueOnce([ + { id: 'run-0-8a521c10-92b1-4013-a4a7-71f20b1a4a4a', data: {} }, + ]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail.mockResolvedValueOnce( + buildRunningWorkflowRun({ + steps: [{ id: 'step-1', type: 'CODE', nextStepIds: [] }], + stepInfos: { 'step-1': { status: StepStatus.RUNNING } }, + }), + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect(mockMetricsService.incrementCounterForEvent).toHaveBeenCalledWith( + expect.objectContaining({ + key: MetricsKeys.WorkflowRunStuckRunningFalsePositive, + }), + ); + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + {}, + ); + expectNoWorkflowRunEnded(); + }); + + it('should keep checking remaining runs when one check fails', async () => { + mockRepository.find.mockResolvedValueOnce([ + { id: 'run-0' }, + { id: 'run-1' }, + ]); + mockWorkflowRunWorkspaceService.getWorkflowRunOrFail + .mockRejectedValueOnce(new Error('boom')) + .mockResolvedValueOnce( + buildRunningWorkflowRun({ + id: 'run-1', + steps: [{ id: 'step-1', type: 'CODE', nextStepIds: [] }], + stepInfos: { 'step-1': { status: StepStatus.RUNNING } }, + }), + ); + + await service.handleStuckRunningRunsForWorkspace(workspaceId); + + expect(mockCacheStorageService.set).toHaveBeenCalledWith( + expect.any(String), + { 'run-1': expect.any(String) }, + ); + expectNoWorkflowRunEnded(); + }); + }); }); diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts index d43c3be256..b3d15ffec9 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts @@ -1,14 +1,31 @@ import { Injectable, Logger } from '@nestjs/common'; import { QUERY_MAX_RECORDS } from 'twenty-shared/constants'; +import { isDefined } from 'twenty-shared/utils'; +import { StepStatus } from 'twenty-shared/workflow'; +import { type FindOptionsWhere } from 'typeorm'; +import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; +import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; +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 { 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 { WorkflowRunStatus, WorkflowRunWorkspaceEntity, } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { workflowHasRunningSteps } from 'src/modules/workflow/common/utils/workflow-has-running-steps.util'; +import { workflowShouldFail } from 'src/modules/workflow/workflow-executor/utils/workflow-should-fail.util'; +import { workflowShouldKeepRunning } from 'src/modules/workflow/workflow-executor/utils/workflow-should-keep-running.util'; +import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; import { getStaledRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.util'; +import { getStuckRunningRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-find-options.util'; +import { getStuckRunningRunsMonitorCacheKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-running-runs-monitor-cache-key.util'; import { getStuckStoppingRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-stopping-runs-find-options.util'; import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; @@ -22,6 +39,11 @@ export class WorkflowHandleStaledRunsWorkspaceService { private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager, private readonly workflowThrottlingWorkspaceService: WorkflowThrottlingWorkspaceService, private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, + private readonly metricsService: MetricsService, + @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) + private readonly cacheStorageService: CacheStorageService, ) {} async handleStaledRunsForWorkspace(workspaceId: string) { @@ -77,39 +99,10 @@ export class WorkflowHandleStaledRunsWorkspaceService { } async handleStuckStoppingRunsForWorkspace(workspaceId: string) { - const authContext = buildSystemAuthContext(workspaceId); - - const stuckStoppingRunIds = - await this.globalWorkspaceOrmManager.executeInWorkspaceContext( - async () => { - const workflowRunRepository = - await this.globalWorkspaceOrmManager.getRepository( - workspaceId, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); - - const findOptions = getStuckStoppingRunsFindOptions(); - - const runIds: string[] = []; - let page: WorkflowRunWorkspaceEntity[]; - - do { - page = await workflowRunRepository.find({ - where: findOptions, - select: { id: true }, - order: { createdAt: 'ASC', id: 'ASC' }, - take: QUERY_MAX_RECORDS, - skip: runIds.length, - }); - - runIds.push(...page.map((workflowRun) => workflowRun.id)); - } while (page.length === QUERY_MAX_RECORDS); - - return runIds; - }, - authContext, - ); + const stuckStoppingRunIds = await this.collectRunIds({ + workspaceId, + findOptions: getStuckStoppingRunsFindOptions(), + }); for (const workflowRunId of stuckStoppingRunIds) { try { @@ -126,4 +119,194 @@ export class WorkflowHandleStaledRunsWorkspaceService { } } } + + // Monitoring mode: stuck RUNNING runs are only flagged, never finalized. + // Flagged runs are re-checked on every sweep; one that ends or gets a new + // job on its own is a false positive, disproving that it was stuck forever. + async handleStuckRunningRunsForWorkspace(workspaceId: string) { + const cacheKey = getStuckRunningRunsMonitorCacheKey(workspaceId); + const flaggedRuns = + (await this.cacheStorageService.get>(cacheKey)) ?? + {}; + + const stuckRunningRunIds = await this.collectRunIds({ + workspaceId, + findOptions: getStuckRunningRunsFindOptions(), + }); + + if ( + stuckRunningRunIds.length === 0 && + Object.keys(flaggedRuns).length === 0 + ) { + return; + } + + const inFlightJobs = + await this.messageQueueService.getInFlightJobs(); + + const hasInFlightJob = (workflowRunId: string) => + inFlightJobs.some( + (job) => + job.id?.startsWith(`${workflowRunId}-`) || + job.data?.workflowRunId === workflowRunId, + ); + + const stillFlaggedRuns: Record = {}; + + for (const [workflowRunId, detectedAt] of Object.entries(flaggedRuns)) { + try { + const workflowRun = + await this.workflowRunWorkspaceService.getWorkflowRunOrFail({ + workflowRunId, + workspaceId, + }); + + if ( + workflowRun.status !== WorkflowRunStatus.RUNNING || + hasInFlightJob(workflowRunId) + ) { + this.logger.warn( + `Stuck running workflow run ${workflowRunId} in workspace ${workspaceId} was a false positive: flagged at ${detectedAt}, now in status ${workflowRun.status}`, + ); + + await this.metricsService.incrementCounterForEvent({ + key: MetricsKeys.WorkflowRunStuckRunningFalsePositive, + eventId: workflowRunId, + }); + } else { + stillFlaggedRuns[workflowRunId] = detectedAt; + } + } catch (error) { + stillFlaggedRuns[workflowRunId] = detectedAt; + this.logger.error( + `Failed to re-check flagged stuck running workflow run ${workflowRunId} for workspace ${workspaceId}`, + error, + ); + } + } + + for (const workflowRunId of stuckRunningRunIds) { + if ( + isDefined(flaggedRuns[workflowRunId]) || + hasInFlightJob(workflowRunId) + ) { + continue; + } + + try { + const expectedOutcome = await this.computeStuckRunningRunOutcome({ + workflowRunId, + workspaceId, + }); + + if (!isDefined(expectedOutcome)) { + continue; + } + + stillFlaggedRuns[workflowRunId] = new Date().toISOString(); + + this.logger.warn( + `Workflow run ${workflowRunId} in workspace ${workspaceId} is stuck in RUNNING without a queue job and would have been finalized as ${expectedOutcome}`, + ); + + await this.metricsService.incrementCounterForEvent({ + key: MetricsKeys.WorkflowRunStuckRunningDetected, + eventId: workflowRunId, + }); + } catch (error) { + this.logger.error( + `Failed to check stuck running workflow run ${workflowRunId} for workspace ${workspaceId}`, + error, + ); + } + } + + await this.cacheStorageService.set(cacheKey, stillFlaggedRuns); + } + + private async computeStuckRunningRunOutcome({ + workflowRunId, + workspaceId, + }: { + workflowRunId: string; + workspaceId: string; + }): Promise { + const workflowRun = + await this.workflowRunWorkspaceService.getWorkflowRunOrFail({ + workflowRunId, + workspaceId, + }); + + if (workflowRun.status !== WorkflowRunStatus.RUNNING) { + return undefined; + } + + const stepInfos = workflowRun.state?.stepInfos; + const steps = workflowRun.state?.flow?.steps; + + if ( + !isDefined(stepInfos) || + !isDefined(steps) || + workflowHasRunningSteps({ stepInfos, steps }) + ) { + return WorkflowRunStatus.FAILED; + } + + if (workflowShouldFail({ stepInfos, steps })) { + return WorkflowRunStatus.FAILED; + } + + const hasPendingSteps = steps.some( + (step) => stepInfos[step.id]?.status === StepStatus.PENDING, + ); + + if (hasPendingSteps) { + return undefined; + } + + if (workflowShouldKeepRunning({ stepInfos, steps })) { + return WorkflowRunStatus.FAILED; + } + + return WorkflowRunStatus.COMPLETED; + } + + private async collectRunIds({ + workspaceId, + findOptions, + }: { + workspaceId: string; + findOptions: FindOptionsWhere; + }): Promise { + const authContext = buildSystemAuthContext(workspaceId); + + return this.globalWorkspaceOrmManager.executeInWorkspaceContext( + async () => { + const workflowRunRepository = + await this.globalWorkspaceOrmManager.getRepository( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const runIds: string[] = []; + let page: WorkflowRunWorkspaceEntity[]; + + do { + page = await workflowRunRepository.find({ + where: findOptions, + select: { id: true }, + order: { createdAt: 'ASC', id: 'ASC' }, + take: QUERY_MAX_RECORDS, + skip: runIds.length, + }); + + runIds.push(...page.map((workflowRun) => workflowRun.id)); + } while (page.length === QUERY_MAX_RECORDS); + + return runIds; + }, + authContext, + ); + } } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts index 84a2edd51c..c1077b4e42 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-enqueue.workspace-service.ts @@ -15,6 +15,7 @@ import { } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; import { RunWorkflowJob } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; +import { buildRunWorkflowJobOptions } from 'src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util'; import { NOT_STARTED_RUNS_FIND_OPTIONS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/not-started-runs-find-options'; import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service'; @@ -117,6 +118,7 @@ export class WorkflowRunEnqueueWorkspaceService { workflowRunId, workspaceId, }, + buildRunWorkflowJobOptions(workflowRunId), ); } diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts index 962322c2ed..5681e78f2b 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts @@ -28,6 +28,7 @@ import { import { RunWorkflowJob } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type'; import { buildRetryStepInfos } from 'src/modules/workflow/workflow-runner/utils/build-retry-step-infos.util'; +import { buildRunWorkflowJobOptions } from 'src/modules/workflow/workflow-runner/utils/build-run-workflow-job-options.util'; import { getRunnableStepIds } from 'src/modules/workflow/workflow-runner/utils/get-runnable-step-ids.util'; import { WorkflowRunEnqueueJob, @@ -130,6 +131,7 @@ export class WorkflowRunnerWorkspaceService { workflowRunId, lastExecutedStepId, }, + buildRunWorkflowJobOptions(workflowRunId), ); } @@ -350,6 +352,7 @@ export class WorkflowRunnerWorkspaceService { workflowRunId, stepIdsToRetry: stepIdsToRun, }, + buildRunWorkflowJobOptions(workflowRunId), ); } catch (error) { // The job couldn't be enqueued: revert to the previous failed state so @@ -446,6 +449,7 @@ export class WorkflowRunnerWorkspaceService { workspaceId, workflowRunId, }, + buildRunWorkflowJobOptions(workflowRunId), ); return { workflowRunId };