fix(server): finalize workflow runs stuck in STOPPING (#22900)

## Context

A workflow run only reaches STOPPED via the in-flight worker execution:
`stopWorkflowRun` just flips a RUNNING run to `STOPPING` (records
intent), and the `STOPPING -> STOPPED` transition is done later inside
`computeWorkflowRunStatus`, which only runs while a worker is executing
the run's steps.

If no worker is executing the run at that point, nothing ever finalizes
it:
- the worker that owned the run crashed / was killed mid-step (e.g.
under heavy load), or
- a step legitimately sits in RUNNING awaiting an external event that
never arrives (the user stopped it).

Only `ENQUEUED` runs had a staleness sweep, so `STOPPING` (and
`RUNNING`) had no recovery path and would stay stuck indefinitely. This
has been observed in production (~150 runs stuck in `STOPPING` after
manual stops during a migration).

## Change

Extend the existing staled-runs machinery to also finalize runs left in
`STOPPING`:
- New `stuck-stopping-runs-threshold` (1h) +
`getStuckStoppingRunsFindOptions` matching `status = STOPPING AND
updatedAt < now - 1h`. `updatedAt` is a TypeORM update-date column, so
it reliably marks when the run entered `STOPPING`, and 1h stays above
any legitimate in-flight step.
- `handleStuckStoppingRunsForWorkspace` finalizes each match to
`STOPPED` via `endWorkflowRun`, so `endedAt`, step infos and the
`WorkflowRunStopped` metric stay consistent. It pages the backlog with
keyset pagination on `(createdAt, id)`, so a page whose finalizations
all fail can't stay at the front of the query and starve later runs
(failed ones are retried on the next sweep).
- Wired into the same cron (`WorkflowHandleStaledRunsCronJob`, every 10
min), per-workspace job, and the manual `workflow:handle-staled-runs`
command — so ops can also clear an existing backlog immediately. The
staled-ENQUEUED and stuck-STOPPING handlers run independently
(`Promise.allSettled` in the job, separate try/catch in the command), so
a failure in one doesn't block the other.

Stop remains manual and unchanged; this only guarantees a stopped run
eventually reaches `STOPPED`.

## Notes / scope

- No schema change (reuses `updatedAt`), so no migration.
- `RUNNING` runs orphaned by a worker crash have the same missing-net
problem; left out of scope here (this covers the user-triggered STOPPING
case).
- The new detection query scans `status`/`updatedAt` like the existing
ENQUEUED sweep; at very high `workflowRun` volumes an index on `(status,
updatedAt)` would help — same pre-existing consideration as the ENQUEUED
path.

## Tests

Unit tests for `handleStuckStoppingRunsForWorkspace`: no-op when none,
finalizes each match to STOPPED, pages through a multi-page backlog, and
advances past a fully-failed page instead of starving later runs. Plus a
unit test for the `(createdAt, id)` keyset condition in
`getStuckStoppingRunsFindOptions`. Full suite green, lint + typecheck
clean on changed files.

Manually verified on a real instance (Postgres): seeded a `STOPPING` run
aged 2h and ran `workflow:handle-staled-runs` -> transitioned to
`STOPPED` with `endedAt` set; a freshly-`STOPPING` run (updatedAt now)
was correctly left untouched by the 1h threshold.

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/22900?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command.ts

@@ -49,6 +49,10 @@ export class WorkflowHandleStaledRunsCommand extends CommandRunner {
         await this.workflowHandleStaledRunsWorkspaceService.handleStaledRunsForWorkspace(
           workspaceId,
         );
+
+        await this.workflowHandleStaledRunsWorkspaceService.handleStuckStoppingRunsForWorkspace(
+          workspaceId,
+        );
       } catch (error) {
         this.logger.error(
           `Failed to handle staled runs for workspace ${workspaceId}`,

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-stopping-runs-threshold.ts

@@ -0,0 +1 @@
+export const STUCK_STOPPING_RUNS_THRESHOLD_MS = 60 * 60 * 1000; // 1 hour

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.cron.job.ts

@@ -18,6 +18,7 @@ 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_STOPPING_RUNS_THRESHOLD_MS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-stopping-runs-threshold';
 import {
   WorkflowHandleStaledRunsJob,
   WorkflowHandleStaledRunsJobData,
@@ -100,9 +101,12 @@ export class WorkflowHandleStaledRunsCronJob {
   }
 
   private async checkAndEnqueue(workspaceId: string): Promise<boolean> {
-    const hasStaledRuns = await this.hasStaledRuns(workspaceId);
+    const [hasStaledRuns, hasStuckStoppingRuns] = await Promise.all([
+      this.hasStaledRuns(workspaceId),
+      this.hasStuckStoppingRuns(workspaceId),
+    ]);
 
-    if (hasStaledRuns) {
+    if (hasStaledRuns || hasStuckStoppingRuns) {
       await this.messageQueueService.add<WorkflowHandleStaledRunsJobData>(
         WorkflowHandleStaledRunsJob.name,
         { workspaceId },
@@ -140,4 +144,18 @@ export class WorkflowHandleStaledRunsCronJob {
 
     return result.length > 0;
   }
+
+  private async hasStuckStoppingRuns(workspaceId: string): Promise<boolean> {
+    const schemaName = getWorkspaceSchemaName(workspaceId);
+    const thresholdDate = new Date(
+      Date.now() - STUCK_STOPPING_RUNS_THRESHOLD_MS,
+    );
+
+    const result = await this.coreDataSource.query(
+      `SELECT 1 FROM ${schemaName}."workflowRun" WHERE "status" = $1 AND "updatedAt" < $2 LIMIT 1`,
+      [WorkflowRunStatus.STOPPING, thresholdDate],
+    );
+
+    return result.length > 0;
+  }
 }

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/jobs/workflow-handle-staled-runs.job.ts

@@ -19,8 +19,13 @@ export class WorkflowHandleStaledRunsJob {
   async handle({
     workspaceId,
   }: WorkflowHandleStaledRunsJobData): Promise<void> {
-    await this.workflowHandleStaledRunsWorkspaceService.handleStaledRunsForWorkspace(
-      workspaceId,
-    );
+    await Promise.all([
+      this.workflowHandleStaledRunsWorkspaceService.handleStaledRunsForWorkspace(
+        workspaceId,
+      ),
+      this.workflowHandleStaledRunsWorkspaceService.handleStuckStoppingRunsForWorkspace(
+        workspaceId,
+      ),
+    ]);
   }
 }

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/__tests__/get-stuck-stopping-runs-find-options.util.spec.ts

@@ -0,0 +1,13 @@
+import { FindOperator } from 'typeorm';
+
+import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
+import { getStuckStoppingRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-stopping-runs-find-options.util';
+
+describe('getStuckStoppingRunsFindOptions', () => {
+  it('should match STOPPING runs older than the threshold', () => {
+    const where = getStuckStoppingRunsFindOptions();
+
+    expect(where.status).toBe(WorkflowRunStatus.STOPPING);
+    expect(where.updatedAt).toBeInstanceOf(FindOperator);
+  });
+});

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-stuck-stopping-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_STOPPING_RUNS_THRESHOLD_MS } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/stuck-stopping-runs-threshold';
+
+export const getStuckStoppingRunsFindOptions =
+  (): FindOptionsWhere<WorkflowRunWorkspaceEntity> => {
+    const thresholdDate = new Date(
+      Date.now() - STUCK_STOPPING_RUNS_THRESHOLD_MS,
+    );
+
+    return {
+      status: WorkflowRunStatus.STOPPING,
+      updatedAt: LessThan(thresholdDate.toISOString()),
+    };
+  };

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts

@@ -7,6 +7,7 @@ import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module';
 import { ThrottlerModule } from 'src/engine/core-modules/throttler/throttler.module';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
+import { WorkflowRunModule } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.module';
 import { WorkflowHandleStaledRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/command/workflow-handle-staled-runs.command';
 import { WorkflowCleanWorkflowRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-clean-workflow-runs.cron.command';
 import { WorkflowHandleStaledRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command';
@@ -29,6 +30,7 @@ import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflo
     WorkspaceDataSourceModule,
     MetricsModule,
     ThrottlerModule,
+    WorkflowRunModule,
   ],
   providers: [
     WorkflowThrottlingWorkspaceService,

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/__tests__/workflow-handle-staled-runs.workspace-service.spec.ts

@@ -4,6 +4,7 @@ import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspac
 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';
 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';
 
 const mockRepository = {
   count: jest.fn(),
@@ -23,13 +24,19 @@ const mockWorkflowThrottlingWorkspaceService = {
   recomputeWorkflowRunNotStartedCount: jest.fn().mockResolvedValue(undefined),
 };
 
+const mockWorkflowRunWorkspaceService = {
+  endWorkflowRun: jest.fn().mockResolvedValue(undefined),
+};
+
 // Mirrors QUERY_MAX_RECORDS, the per-batch cap the service applies. Kept as a
 // local literal rather than imported from twenty-shared/constants to avoid a
 // jest circular-init issue with config-variables loading the same barrel.
 const QUERY_MAX_RECORDS = 200;
 
-const buildStaledRuns = (count: number) =>
-  Array.from({ length: count }, (_, index) => ({ id: `run-${index}` }));
+const buildStaledRuns = (count: number, startIndex = 0) =>
+  Array.from({ length: count }, (_, index) => ({
+    id: `run-${startIndex + index}`,
+  }));
 
 describe('WorkflowHandleStaledRunsWorkspaceService', () => {
   let service: WorkflowHandleStaledRunsWorkspaceService;
@@ -50,6 +57,10 @@ describe('WorkflowHandleStaledRunsWorkspaceService', () => {
           provide: WorkflowThrottlingWorkspaceService,
           useValue: mockWorkflowThrottlingWorkspaceService,
         },
+        {
+          provide: WorkflowRunWorkspaceService,
+          useValue: mockWorkflowRunWorkspaceService,
+        },
       ],
     }).compile();
 
@@ -135,4 +146,95 @@ describe('WorkflowHandleStaledRunsWorkspaceService', () => {
       mockWorkflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount,
     ).toHaveBeenCalledTimes(1);
   });
+
+  describe('handleStuckStoppingRunsForWorkspace', () => {
+    it('should do nothing when there are no stuck stopping runs', async () => {
+      mockRepository.find.mockResolvedValueOnce([]);
+
+      await service.handleStuckStoppingRunsForWorkspace(workspaceId);
+
+      expect(
+        mockWorkflowRunWorkspaceService.endWorkflowRun,
+      ).not.toHaveBeenCalled();
+    });
+
+    it('should finalize each stuck stopping run to STOPPED', async () => {
+      mockRepository.find.mockResolvedValueOnce(buildStaledRuns(3));
+
+      await service.handleStuckStoppingRunsForWorkspace(workspaceId);
+
+      expect(
+        mockWorkflowRunWorkspaceService.endWorkflowRun,
+      ).toHaveBeenCalledTimes(3);
+      expect(
+        mockWorkflowRunWorkspaceService.endWorkflowRun,
+      ).toHaveBeenCalledWith({
+        workflowRunId: 'run-0',
+        workspaceId,
+        status: WorkflowRunStatus.STOPPED,
+      });
+    });
+
+    it('should page through a multi-page backlog until a short page', async () => {
+      mockRepository.find
+        .mockResolvedValueOnce(buildStaledRuns(QUERY_MAX_RECORDS))
+        .mockResolvedValueOnce(
+          buildStaledRuns(QUERY_MAX_RECORDS, QUERY_MAX_RECORDS),
+        )
+        .mockResolvedValueOnce(buildStaledRuns(50, 2 * QUERY_MAX_RECORDS));
+
+      await service.handleStuckStoppingRunsForWorkspace(workspaceId);
+
+      expect(mockRepository.find).toHaveBeenCalledTimes(3);
+      expect(mockRepository.find).toHaveBeenCalledWith(
+        expect.objectContaining({ take: QUERY_MAX_RECORDS }),
+      );
+      expect(
+        mockWorkflowRunWorkspaceService.endWorkflowRun,
+      ).toHaveBeenCalledTimes(2 * QUERY_MAX_RECORDS + 50);
+    });
+
+    it('should keep finalizing remaining runs when one fails', async () => {
+      mockRepository.find.mockResolvedValueOnce(buildStaledRuns(3));
+      mockWorkflowRunWorkspaceService.endWorkflowRun
+        .mockRejectedValueOnce(new Error('boom'))
+        .mockResolvedValue(undefined);
+
+      await service.handleStuckStoppingRunsForWorkspace(workspaceId);
+
+      expect(
+        mockWorkflowRunWorkspaceService.endWorkflowRun,
+      ).toHaveBeenCalledTimes(3);
+    });
+
+    it('should advance past a fully failed page instead of starving later runs', async () => {
+      mockRepository.find
+        .mockResolvedValueOnce(buildStaledRuns(QUERY_MAX_RECORDS))
+        .mockResolvedValueOnce(buildStaledRuns(50, QUERY_MAX_RECORDS));
+      // Whole first page fails, later runs still get finalized
+      mockWorkflowRunWorkspaceService.endWorkflowRun.mockImplementation(
+        ({ workflowRunId }: { workflowRunId: string }) => {
+          const index = Number(workflowRunId.replace('run-', ''));
+
+          return index < QUERY_MAX_RECORDS
+            ? Promise.reject(new Error('boom'))
+            : Promise.resolve(undefined);
+        },
+      );
+
+      await service.handleStuckStoppingRunsForWorkspace(workspaceId);
+
+      expect(mockRepository.find).toHaveBeenCalledTimes(2);
+      expect(
+        mockWorkflowRunWorkspaceService.endWorkflowRun,
+      ).toHaveBeenCalledTimes(QUERY_MAX_RECORDS + 50);
+      expect(
+        mockWorkflowRunWorkspaceService.endWorkflowRun,
+      ).toHaveBeenCalledWith({
+        workflowRunId: `run-${QUERY_MAX_RECORDS}`,
+        workspaceId,
+        status: WorkflowRunStatus.STOPPED,
+      });
+    });
+  });
 });

packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-handle-staled-runs.workspace-service.ts

@@ -9,7 +9,9 @@ import {
   WorkflowRunWorkspaceEntity,
 } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
 import { getStaledRunsFindOptions } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-staled-runs-find-options.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';
 
 @Injectable()
 export class WorkflowHandleStaledRunsWorkspaceService {
@@ -19,6 +21,7 @@ export class WorkflowHandleStaledRunsWorkspaceService {
   constructor(
     private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
     private readonly workflowThrottlingWorkspaceService: WorkflowThrottlingWorkspaceService,
+    private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService,
   ) {}
 
   async handleStaledRunsForWorkspace(workspaceId: string) {
@@ -72,4 +75,55 @@ export class WorkflowHandleStaledRunsWorkspaceService {
       );
     }, authContext);
   }
+
+  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,
+      );
+
+    for (const workflowRunId of stuckStoppingRunIds) {
+      try {
+        await this.workflowRunWorkspaceService.endWorkflowRun({
+          workflowRunId,
+          workspaceId,
+          status: WorkflowRunStatus.STOPPED,
+        });
+      } catch (error) {
+        this.logger.error(
+          `Failed to finalize stuck stopping workflow run ${workflowRunId} for workspace ${workspaceId}`,
+          error,
+        );
+      }
+    }
+  }
 }