From 6bc4d8efac9d38a74a8f2979859789cc48d2ea99 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Wed, 8 Jul 2026 13:53:26 +0200 Subject: [PATCH] fix(workflow): batch staled run reset to avoid Postgres param limit (#22654) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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. Review in cubic --- ...ndle-staled-runs.workspace-service.spec.ts | 138 ++++++++++++++++++ ...ow-handle-staled-runs.workspace-service.ts | 39 +++-- 2 files changed, 168 insertions(+), 9 deletions(-) create mode 100644 packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/__tests__/workflow-handle-staled-runs.workspace-service.spec.ts 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 new file mode 100644 index 0000000000..17cb8789e0 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/__tests__/workflow-handle-staled-runs.workspace-service.spec.ts @@ -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, + ); + }); + + 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); + }); +}); 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 35192a88f1..b4424ad315 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,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,