fix(server): recover workflow runs whose queue job was lost - monitoring only (#22995)

## 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 `<workflowRunId>-<uuid>`. Worker logs (`Processing job <id>` / `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.
2026-07-20 14:21:33 +02:00
parent 87729a2822
commit 98c71d7b3d
21 changed files with 720 additions and 39 deletions
@@ -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<T>({
@@ -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<T extends MessageQueueJobData>(
+    queueName: MessageQueue,
+  ): Promise<InFlightQueueJob<T>[]> {
+    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 }));
+  }
 }
@@ -1,5 +1,6 @@
 export interface QueueJobOptions {
  id?: string;
+  allowDuplicatedPrefixes?: boolean;
  priority?: number;
  retryLimit?: number;
  delay?: number;
@@ -42,4 +42,12 @@ export interface MessageQueueDriver {
    jobId?: string;
  }): Promise<void>;
  register?(queueName: MessageQueue): void;
+  getInFlightJobs?<T extends MessageQueueJobData>(
+    queueName: MessageQueue,
+  ): Promise<InFlightQueueJob<T>[]>;
+}
+
+export interface InFlightQueueJob<T extends MessageQueueJobData> {
+  id?: string;
+  data: T;
 }
@@ -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<T extends MessageQueueJobData>(): Promise<
+    InFlightQueueJob<T>[]
+  > {
+    if (typeof this.driver.getInFlightJobs !== 'function') {
+      return Promise.resolve([]);
+    }
+
+    return this.driver.getInFlightJobs(this.queueName);
+  }
+
  addCron<T extends MessageQueueJobData | undefined>({
    jobName,
    data,
@@ -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',