fix(workflow): batch staled run reset to avoid Postgres param limit (#22654)

## Problem Self-hosters with a large backlog of workflow runs stuck in `ENQUEUED` see the recovery job (`WorkflowHandleStaledRunsJob`) fail with: ``` Error: Data validation error. at computeTwentyORMException ... at WorkspaceSelectQueryBuilder.getMany ... at WorkspaceUpdateQueryBuilder.execute ... at WorkflowHandleStaledRunsWorkspaceService.handleStaledRunsForWorkspace ... ``` So the very job meant to unblock enqueued runs can never complete, and runs stay stuck. ## Root cause `handleStaledRunsForWorkspace` fetched **every** staled run unbounded, then called `repository.update(allIds, ...)`. That builds a `WHERE id IN ($1, $2, ... $N)`. Inside `WorkspaceUpdateQueryBuilder.execute`, a "before" `SELECT` runs with that same huge `IN` list; with a big enough backlog the bind-parameter count exceeds Postgres' limit, the `getMany` throws a `QueryFailedError`, and `computeTwentyORMException` maps the resulting PG error code to the generic `PostgresException('Data validation error.')`. There's also a secondary `before.length > QUERY_MAX_RECORDS` (200) guard in the update path that would reject anything over 200 rows even if the param limit weren't hit. ## Fix Process staled runs in batches of `QUERY_MAX_RECORDS` (200), looping until a pass finds none left — the same batching pattern the sibling clean-runs job already uses. Each update flips the batch from `ENQUEUED` to `NOT_STARTED`, so the find criteria stops matching them and the loop terminates. The throttling recompute now runs once at the end, and only if at least one batch was reset. ## Tests New unit spec covering: - no staled runs -> no update, no recompute - single batch -> correct ids/payload, recompute once - exactly 200 -> `take: 200`, 200 ids per update - 450 backlog -> 3 update calls (200/200/50), loops until empty, recompute exactly once All 4 pass locally. ## Note This fixes the recovery job. If runs keep re-accumulating as `ENQUEUED`, there may be a separate producer-side issue worth investigating.  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22654?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>
2026-07-08 13:53:26 +02:00
parent bfeaaa56a3
commit 6bc4d8efac
2 changed files with 168 additions and 9 deletions
@@ -0,0 +1,138 @@
+import { Test, type TestingModule } from '@nestjs/testing';
+
+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';
+import { WorkflowThrottlingWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-throttling.workspace-service';
+
+const mockRepository = {
+  count: jest.fn(),
+  find: jest.fn(),
+  update: jest.fn().mockResolvedValue(undefined),
+};
+
+const mockGlobalWorkspaceOrmManager = {
+  getRepository: jest.fn().mockResolvedValue(mockRepository),
+  executeInWorkspaceContext: jest
+    .fn()
+    // oxlint-disable-next-line typescript/no-explicit-any
+    .mockImplementation((fn: () => any) => fn()),
+};
+
+const mockWorkflowThrottlingWorkspaceService = {
+  recomputeWorkflowRunNotStartedCount: 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}` }));
+
+describe('WorkflowHandleStaledRunsWorkspaceService', () => {
+  let service: WorkflowHandleStaledRunsWorkspaceService;
+
+  const workspaceId = 'workspace-1';
+
+  beforeEach(async () => {
+    jest.clearAllMocks();
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        WorkflowHandleStaledRunsWorkspaceService,
+        {
+          provide: GlobalWorkspaceOrmManager,
+          useValue: mockGlobalWorkspaceOrmManager,
+        },
+        {
+          provide: WorkflowThrottlingWorkspaceService,
+          useValue: mockWorkflowThrottlingWorkspaceService,
+        },
+      ],
+    }).compile();
+
+    service = module.get<WorkflowHandleStaledRunsWorkspaceService>(
+      WorkflowHandleStaledRunsWorkspaceService,
+    );
+  });
+
+  it('should do nothing when there are no staled runs', async () => {
+    mockRepository.count.mockResolvedValueOnce(0);
+
+    await service.handleStaledRunsForWorkspace(workspaceId);
+
+    expect(mockRepository.find).not.toHaveBeenCalled();
+    expect(mockRepository.update).not.toHaveBeenCalled();
+    expect(
+      mockWorkflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount,
+    ).not.toHaveBeenCalled();
+  });
+
+  it('should reset a single batch and recompute the not-started count once', async () => {
+    const staledRuns = buildStaledRuns(3);
+
+    mockRepository.count.mockResolvedValueOnce(3);
+    mockRepository.find.mockResolvedValueOnce(staledRuns);
+
+    await service.handleStaledRunsForWorkspace(workspaceId);
+
+    expect(mockRepository.update).toHaveBeenCalledTimes(1);
+    expect(mockRepository.update).toHaveBeenCalledWith(
+      ['run-0', 'run-1', 'run-2'],
+      {
+        enqueuedAt: null,
+        status: WorkflowRunStatus.NOT_STARTED,
+      },
+    );
+    expect(
+      mockWorkflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount,
+    ).toHaveBeenCalledTimes(1);
+    expect(
+      mockWorkflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount,
+    ).toHaveBeenCalledWith(workspaceId);
+  });
+
+  it('should never fetch more than QUERY_MAX_RECORDS ids per update', async () => {
+    mockRepository.count.mockResolvedValueOnce(QUERY_MAX_RECORDS);
+    mockRepository.find.mockResolvedValueOnce(
+      buildStaledRuns(QUERY_MAX_RECORDS),
+    );
+
+    await service.handleStaledRunsForWorkspace(workspaceId);
+
+    expect(mockRepository.find).toHaveBeenCalledWith(
+      expect.objectContaining({ take: QUERY_MAX_RECORDS }),
+    );
+    expect(mockRepository.update.mock.calls[0][0]).toHaveLength(
+      QUERY_MAX_RECORDS,
+    );
+  });
+
+  it('should drain a multi-batch backlog based on the initial count', async () => {
+    // 450 staled runs => 3 batches (200, 200, 50)
+    mockRepository.count.mockResolvedValueOnce(450);
+    mockRepository.find
+      .mockResolvedValueOnce(buildStaledRuns(QUERY_MAX_RECORDS))
+      .mockResolvedValueOnce(buildStaledRuns(QUERY_MAX_RECORDS))
+      .mockResolvedValueOnce(buildStaledRuns(50));
+
+    await service.handleStaledRunsForWorkspace(workspaceId);
+
+    expect(mockRepository.find).toHaveBeenCalledTimes(3);
+    expect(mockRepository.update).toHaveBeenCalledTimes(3);
+    expect(mockRepository.update.mock.calls[0][0]).toHaveLength(
+      QUERY_MAX_RECORDS,
+    );
+    expect(mockRepository.update.mock.calls[1][0]).toHaveLength(
+      QUERY_MAX_RECORDS,
+    );
+    expect(mockRepository.update.mock.calls[2][0]).toHaveLength(50);
+
+    // Recompute runs exactly once, after the whole backlog is drained
+    expect(
+      mockWorkflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount,
+    ).toHaveBeenCalledTimes(1);
+  });
+});
@@ -1,5 +1,7 @@
 import { Injectable, Logger } from '@nestjs/common';

+import { QUERY_MAX_RECORDS } from 'twenty-shared/constants';
+
 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 {
@@ -30,21 +32,40 @@ export class WorkflowHandleStaledRunsWorkspaceService {
          { shouldBypassPermissionChecks: true },
        );

-      const staledWorkflowRuns = await workflowRunRepository.find({
+      const staledRunsCount = await workflowRunRepository.count({
        where: getStaledRunsFindOptions(),
      });

-      if (staledWorkflowRuns.length <= 0) {
+      if (staledRunsCount <= 0) {
        return;
      }

-      await workflowRunRepository.update(
-        staledWorkflowRuns.map((workflowRun) => workflowRun.id),
-        {
-          enqueuedAt: null,
-          status: WorkflowRunStatus.NOT_STARTED,
-        },
-      );
+      const batchCount = Math.ceil(staledRunsCount / QUERY_MAX_RECORDS);
+
+      for (let batchIndex = 0; batchIndex < batchCount; batchIndex++) {
+        const staledWorkflowRuns = await workflowRunRepository.find({
+          where: getStaledRunsFindOptions(),
+          select: {
+            id: true,
+          },
+          order: {
+            createdAt: 'ASC',
+          },
+          take: QUERY_MAX_RECORDS,
+        });
+
+        if (staledWorkflowRuns.length <= 0) {
+          break;
+        }
+
+        await workflowRunRepository.update(
+          staledWorkflowRuns.map((workflowRun) => workflowRun.id),
+          {
+            enqueuedAt: null,
+            status: WorkflowRunStatus.NOT_STARTED,
+          },
+        );
+      }

      await this.workflowThrottlingWorkspaceService.recomputeWorkflowRunNotStartedCount(
        workspaceId,