From 0b809c987a4df86155e98a9f0e2ac52257c8fc16 Mon Sep 17 00:00:00 2001 From: Jan Kalkan Date: Thu, 16 Jul 2026 21:20:13 +0200 Subject: [PATCH] fix(core): Delete executions in batches when removing workflows, projects and folders (#34265) Co-authored-by: Claude Fable 5 --- .../__tests__/execution-persistence.test.ts | 109 +++++++++++++++++ .../src/executions/execution-persistence.ts | 98 +++++++++++++++ .../source-control-import.service.ee.test.ts | 81 +++++++++++++ .../source-control-import.service.ee.ts | 112 ++++++++++++++++-- .../__tests__/workflow.service.test.ts | 32 +++-- .../cli/src/workflows/workflow.service.ts | 19 +-- .../source-control-import.service.test.ts | 2 + .../workflows/workflow.service.test.ts | 3 +- 8 files changed, 421 insertions(+), 35 deletions(-) diff --git a/packages/cli/src/executions/__tests__/execution-persistence.test.ts b/packages/cli/src/executions/__tests__/execution-persistence.test.ts index e00c8d44f2b..23ae6a80878 100644 --- a/packages/cli/src/executions/__tests__/execution-persistence.test.ts +++ b/packages/cli/src/executions/__tests__/execution-persistence.test.ts @@ -1855,6 +1855,115 @@ describe('ExecutionPersistence', () => { }); }); + describe('hardDeleteByWorkflowId', () => { + const executionPersistence = createPersistenceService('db'); + + const executionRow = (id: string, storedAt: 'db' | 'fs' = 'db') => + Object.assign(new ExecutionEntity(), { id, workflowId: 'wf-1', storedAt }); + + it('should delete executions in batches until none remain, including soft-deleted ones', async () => { + executionRepository.find + .mockResolvedValueOnce([executionRow('exec-1'), executionRow('exec-2', 'fs')]) + .mockResolvedValueOnce([executionRow('exec-3')]) + .mockResolvedValueOnce([]); + + await executionPersistence.hardDeleteByWorkflowId('wf-1'); + + expect(executionRepository.find).toHaveBeenCalledTimes(3); + expect(executionRepository.find).toHaveBeenCalledWith({ + select: ['id', 'workflowId', 'storedAt'], + where: { workflowId: 'wf-1' }, + take: 500, + withDeleted: true, + }); + expect(executionRepository.deleteByIds).toHaveBeenNthCalledWith(1, ['exec-1', 'exec-2']); + expect(executionRepository.deleteByIds).toHaveBeenNthCalledWith(2, ['exec-3']); + expect(binaryDataService.deleteMany).toHaveBeenCalledTimes(2); + expect(jsonStore.delete).toHaveBeenNthCalledWith(1, [ + { executionId: 'exec-2', workflowId: 'wf-1', storedAt: 'fs' }, + ]); + }); + + it('should delete nothing when the workflow has no executions', async () => { + executionRepository.find.mockResolvedValueOnce([]); + + await executionPersistence.hardDeleteByWorkflowId('wf-1'); + + expect(executionRepository.deleteByIds).not.toHaveBeenCalled(); + expect(binaryDataService.deleteMany).not.toHaveBeenCalled(); + expect(jsonStore.delete).not.toHaveBeenCalled(); + }); + + it('should propagate a batch failure without deleting further batches', async () => { + executionRepository.find + .mockResolvedValueOnce([executionRow('exec-1')]) + .mockResolvedValueOnce([executionRow('exec-2')]); + executionRepository.deleteByIds + .mockResolvedValueOnce(mock()) + .mockRejectedValueOnce(new Error('connection lost')); + + await expect(executionPersistence.hardDeleteByWorkflowId('wf-1')).rejects.toThrow( + 'connection lost', + ); + + // first batch was deleted before the failure; retrying resumes from the rest + expect(executionRepository.deleteByIds).toHaveBeenNthCalledWith(1, ['exec-1']); + expect(executionRepository.deleteByIds).toHaveBeenNthCalledWith(2, ['exec-2']); + expect(executionRepository.find).toHaveBeenCalledTimes(2); + }); + + it('should throw instead of looping on when executions keep being added', async () => { + const sqlitePersistence = createPersistenceService('db', 'sqlite'); + executionRepository.find.mockResolvedValue([executionRow('exec-1')]); + + await expect(sqlitePersistence.hardDeleteByWorkflowId('wf-1')).rejects.toThrow( + 'executions keep being added', + ); + + expect(executionRepository.find).toHaveBeenCalledTimes(20_001); // fixed per-run batch cap + final probe + expect(executionRepository.query).not.toHaveBeenCalled(); // no catalog estimate outside Postgres + executionRepository.find.mockReset(); + }); + + it('should succeed when the deletion converges exactly on the last allowed batch', async () => { + const sqlitePersistence = createPersistenceService('db', 'sqlite'); + for (let i = 0; i < 20_000; i++) { + executionRepository.find.mockResolvedValueOnce([executionRow('exec-1')]); + } + executionRepository.find.mockResolvedValue([]); // final probe finds none left + + await expect(sqlitePersistence.hardDeleteByWorkflowId('wf-1')).resolves.toBeUndefined(); + + expect(executionRepository.find).toHaveBeenCalledTimes(20_001); // 20k full batches + final probe + executionRepository.find.mockReset(); + }); + + it('should scale the safeguard cap with the table-size estimate on Postgres', async () => { + executionRepository.query.mockResolvedValueOnce([{ estimate: '10000000' }]); + executionRepository.find.mockResolvedValue([executionRow('exec-1')]); + + await expect(executionPersistence.hardDeleteByWorkflowId('wf-1')).rejects.toThrow( + 'executions keep being added', + ); + + // 2 x 10M estimate / 500 per batch = 40k batches, + final probe + expect(executionRepository.find).toHaveBeenCalledTimes(40_001); + executionRepository.find.mockReset(); + }); + + it('should fall back to the fixed cap when the Postgres estimate is unavailable', async () => { + executionRepository.query.mockResolvedValueOnce([{ estimate: '-1' }]); // never-analyzed table + executionRepository.find.mockResolvedValue([executionRow('exec-1')]); + + await expect(executionPersistence.hardDeleteByWorkflowId('wf-1')).rejects.toThrow( + 'executions keep being added', + ); + + expect(executionRepository.find).toHaveBeenCalledTimes(20_001); + executionRepository.find.mockReset(); + }); + }); + describe('deleteUnsaved', () => { const target = { workflowId: 'wf-1', executionId: 'exec-1', storedAt: 'db' as const }; diff --git a/packages/cli/src/executions/execution-persistence.ts b/packages/cli/src/executions/execution-persistence.ts index 80c0cfd3c21..8a804a415eb 100644 --- a/packages/cli/src/executions/execution-persistence.ts +++ b/packages/cli/src/executions/execution-persistence.ts @@ -61,6 +61,18 @@ type UpdatableEntityColumns = Omit< */ @Service() export class ExecutionPersistence { + /** Batch size for bulk deletion: stays below SQLite's expression-tree depth limit (~1000). */ + private static readonly bulkDeletionBatchSize = 500; + + /** + * Fallback runaway safeguard: caps one bulk-deletion run at 10M executions. + * On Postgres the cap scales with the table-size estimate instead (see + * `maxBulkDeletionBatches`). Deletion resumes on retry, so a larger history + * still converges across runs — only a workflow that keeps producing + * executions hits the cap repeatedly. + */ + private static readonly maxBulkDeletionBatchesPerRun = 20_000; + constructor( private readonly executionRepository: ExecutionRepository, private readonly binaryDataService: BinaryDataService, @@ -607,6 +619,92 @@ export class ExecutionPersistence { await this.jsonStore.delete(this.toBlobRefs(refs)); } + /** + * Hard-delete all executions of a workflow, in batches small enough that no + * single statement can exceed a DB statement timeout, no matter how large + * the execution history is. Each batch commits independently, so an + * interrupted deletion picks up where it left off when retried. + * + * Callers must deactivate the workflow first — if something keeps producing + * executions for it, this throws once the per-run batch cap is exhausted, + * instead of looping forever. + */ + async hardDeleteByWorkflowId(workflowId: string) { + const maxBatches = await this.maxBulkDeletionBatches(); + + for (let batch = 0; batch < maxBatches; batch++) { + const executions = await this.executionRepository.find({ + select: ['id', 'workflowId', 'storedAt'], + where: { workflowId }, + take: ExecutionPersistence.bulkDeletionBatchSize, + withDeleted: true, // sweep soft-deleted executions too, or they'd be left to the FK cascade + }); + + if (executions.length === 0) return; + + await this.hardDelete( + executions.map((execution) => ({ + executionId: execution.id, + workflowId: execution.workflowId, + storedAt: execution.storedAt, + })), + ); + } + + // The loop may have converged exactly on its last allowed batch - only + // fail if executions actually remain. + const remaining = await this.executionRepository.find({ + select: ['id'], + where: { workflowId }, + take: 1, + withDeleted: true, + }); + if (remaining.length === 0) return; + + throw new UnexpectedError( + `Failed to delete all executions of workflow ${workflowId}: executions keep being added while deleting them - is the workflow still active?`, + ); + } + + /** + * Runaway-safeguard cap for `hardDeleteByWorkflowId`: twice the table-size + * estimate when one is available — no single workflow can have more + * executions than the whole table, so large deployments never hit the cap + * on legitimate work — floored at the fixed per-run cap, which is also the + * fallback when no estimate is available. + */ + private async maxBulkDeletionBatches(): Promise { + const { bulkDeletionBatchSize, maxBulkDeletionBatchesPerRun: fallback } = ExecutionPersistence; + + const estimate = await this.estimateExecutionsTableSize(); + if (estimate === null) return fallback; + + return Math.max(Math.ceil((estimate * 2) / bulkDeletionBatchSize), fallback); + } + + /** + * Table-size estimate for executions from the Postgres system catalogs + * (O(1) lookup). Returns null on other DBs, on never-analyzed tables + * (`reltuples` is -1), or when the lookup fails — this is best-effort only. + */ + private async estimateExecutionsTableSize(): Promise { + if (this.databaseConfig.type !== 'postgresdb') return null; + + try { + const { schema, tableName } = this.executionRepository.metadata; + const table = schema ? `"${schema}"."${tableName}"` : `"${tableName}"`; + const rows = (await this.executionRepository.query( + 'SELECT reltuples::bigint AS estimate FROM pg_class WHERE oid = to_regclass($1)', + [table], + )) as Array<{ estimate: string | number }>; + + const estimate = Number(rows[0]?.estimate); + return Number.isFinite(estimate) && estimate >= 0 ? estimate : null; + } catch { + return null; + } + } + /** Narrow deletion targets to those whose data lives in a blob store, i.e. all but `db`. */ private toBlobRefs(targets: T[]) { return targets.filter((t): t is T & { storedAt: BlobStorageLocation } => t.storedAt !== 'db'); diff --git a/packages/cli/src/modules/source-control.ee/__tests__/source-control-import.service.ee.test.ts b/packages/cli/src/modules/source-control.ee/__tests__/source-control-import.service.ee.test.ts index f396064c283..58ee61360a7 100644 --- a/packages/cli/src/modules/source-control.ee/__tests__/source-control-import.service.ee.test.ts +++ b/packages/cli/src/modules/source-control.ee/__tests__/source-control-import.service.ee.test.ts @@ -12,6 +12,7 @@ import { type ProjectRelation, type ProjectRelationRepository, type ProjectRepository, + type SharedWorkflow, type SharedWorkflowRepository, type TagRepository, type WorkflowTagMappingRepository, @@ -29,7 +30,9 @@ vi.mock('node:fs/promises'); import type { Mock } from 'vitest'; import { mock } from 'vitest-mock-extended'; +import type { ActiveWorkflowManager } from '@/active-workflow-manager'; import type { VariablesService } from '@/environments.ee/variables/variables.service.ee'; +import type { ExecutionPersistence } from '@/executions/execution-persistence'; import type { DataTableColumnRepository } from '@/modules/data-table/data-table-column.repository'; import type { DataTableDDLService } from '@/modules/data-table/data-table-ddl.service'; import type { DataTableSizeValidator } from '@/modules/data-table/data-table-size-validator.service'; @@ -70,6 +73,8 @@ describe('SourceControlImportService', () => { const dataTableDDLService = mock(); const redactionEnforcementService = mock(); const dataTableSizeValidator = mock(); + const activeWorkflowManager = mock(); + const executionPersistence = mock(); const globalAdminContext = new SourceControlContext( Object.assign(new User(), { role: GLOBAL_ADMIN_ROLE }), @@ -109,6 +114,8 @@ describe('SourceControlImportService', () => { dataTableDDLService, redactionEnforcementService, dataTableSizeValidator, + activeWorkflowManager, + executionPersistence, ); const globMock = fastGlob.default as unknown as Mock<(...args: string[]) => Promise>; @@ -2467,6 +2474,31 @@ describe('SourceControlImportService', () => { }); }); + describe('deleteWorkflowsNotInWorkfolder', () => { + it('should wrap deletion failures with resource context', async () => { + const user = Object.assign(new User(), { id: 'user-1' }); + const candidate = mock({ id: 'wf-1', name: 'My workflow' }); + workflowService.delete.mockRejectedValueOnce(new Error('statement timeout')); + + await expect(service.deleteWorkflowsNotInWorkfolder(user, [candidate])).rejects.toThrow( + 'Failed to delete workflow(s) "My workflow" (wf-1) while pulling from source control: statement timeout', + ); + }); + }); + + describe('deletionError resource listing', () => { + it('should cap the number of resources listed in the error message', async () => { + const candidates = Array.from({ length: 12 }, (_, i) => + mock({ id: `project-${i}`, name: `Project ${i}` }), + ); + sharedWorkflowRepository.find.mockRejectedValueOnce(new Error('boom')); + + await expect(service.deleteTeamProjectsNotInWorkfolder(candidates)).rejects.toThrow( + '"Project 9" (project-9) and 2 more while pulling from source control: boom', + ); + }); + }); + describe('deleteFoldersNotInWorkfolder', () => { it('should call folderRepository.delete with correct ids', async () => { const candidates = [ @@ -2474,6 +2506,9 @@ describe('SourceControlImportService', () => { mock({ id: 'folder2' }), mock({ id: 'folder3' }), ]; + folderRepository.getAllFolderIdsInHierarchy.mockResolvedValue([]); + workflowRepository.find.mockResolvedValue([]); + await service.deleteFoldersNotInWorkfolder(candidates as any); expect(folderRepository.delete).toHaveBeenCalledWith({ @@ -2485,6 +2520,24 @@ describe('SourceControlImportService', () => { await service.deleteFoldersNotInWorkfolder([]); expect(folderRepository.delete).not.toHaveBeenCalled(); }); + + it('should drain workflows in the folder hierarchy before deleting folders', async () => { + const candidates = [mock({ id: 'folder1' })]; + const straggler = Object.assign(new WorkflowEntity(), { id: 'wf-1', active: false }); + folderRepository.getAllFolderIdsInHierarchy.mockResolvedValueOnce(['subfolder1']); + workflowRepository.find.mockResolvedValueOnce([straggler]); // workflows in the hierarchy + workflowRepository.findOne.mockResolvedValueOnce(straggler); // active-flag lookup before draining + + await service.deleteFoldersNotInWorkfolder(candidates as any); + + expect(workflowRepository.find).toHaveBeenNthCalledWith(1, { + select: ['id'], + where: { parentFolder: { id: In(['folder1', 'subfolder1']) } }, + }); + expect(activeWorkflowManager.remove).not.toHaveBeenCalled(); + expect(executionPersistence.hardDeleteByWorkflowId).toHaveBeenCalledWith('wf-1'); + expect(folderRepository.delete).toHaveBeenCalledWith({ id: In(['folder1']) }); + }); }); }); @@ -3005,6 +3058,7 @@ describe('SourceControlImportService', () => { mock({ id: 'project-1' }), mock({ id: 'project-2' }), ]; + sharedWorkflowRepository.find.mockResolvedValue([]); await service.deleteTeamProjectsNotInWorkfolder(candidates); @@ -3018,6 +3072,33 @@ describe('SourceControlImportService', () => { expect(projectRepository.delete).not.toHaveBeenCalled(); }); + + it('should deactivate and drain straggler workflows before deleting projects', async () => { + const candidates = [mock({ id: 'project-1' })]; + sharedWorkflowRepository.find.mockResolvedValueOnce([ + { workflowId: 'wf-active' }, + { workflowId: 'wf-inactive' }, + ] as SharedWorkflow[]); + workflowRepository.findOne + .mockResolvedValueOnce( + Object.assign(new WorkflowEntity(), { id: 'wf-active', active: true }), + ) + .mockResolvedValueOnce( + Object.assign(new WorkflowEntity(), { id: 'wf-inactive', active: false }), + ); + + await service.deleteTeamProjectsNotInWorkfolder(candidates); + + expect(sharedWorkflowRepository.find).toHaveBeenCalledWith({ + select: ['workflowId'], + where: { projectId: In(['project-1']), role: 'workflow:owner' }, + }); + expect(activeWorkflowManager.remove).toHaveBeenCalledTimes(1); + expect(activeWorkflowManager.remove).toHaveBeenCalledWith('wf-active'); + expect(executionPersistence.hardDeleteByWorkflowId).toHaveBeenCalledWith('wf-active'); + expect(executionPersistence.hardDeleteByWorkflowId).toHaveBeenCalledWith('wf-inactive'); + expect(projectRepository.delete).toHaveBeenCalledWith({ id: In(['project-1']) }); + }); }); }); diff --git a/packages/cli/src/modules/source-control.ee/source-control-import.service.ee.ts b/packages/cli/src/modules/source-control.ee/source-control-import.service.ee.ts index 024b7b2a94c..c4b209ffe2c 100644 --- a/packages/cli/src/modules/source-control.ee/source-control-import.service.ee.ts +++ b/packages/cli/src/modules/source-control.ee/source-control-import.service.ee.ts @@ -36,7 +36,9 @@ import { shouldAutoPublishWorkflow, jsonParse, UnexpectedError, UserError } from import { readFile as fsReadFile } from 'node:fs/promises'; import path from 'path'; +import { ActiveWorkflowManager } from '@/active-workflow-manager'; import { CredentialsService } from '@/credentials/credentials.service'; +import { ExecutionPersistence } from '@/executions/execution-persistence'; import type { IWorkflowToImport } from '@/interfaces'; import { DataTableColumn } from '@/modules/data-table/data-table-column.entity'; import { DataTableColumnRepository } from '@/modules/data-table/data-table-column.repository'; @@ -145,6 +147,8 @@ export class SourceControlImportService { private readonly dataTableDDLService: DataTableDDLService, private readonly redactionEnforcementService: RedactionEnforcementService, private readonly dataTableSizeValidator: DataTableSizeValidator, + private readonly activeWorkflowManager: ActiveWorkflowManager, + private readonly executionPersistence: ExecutionPersistence, ) { this.gitFolder = path.join(instanceSettings.n8nFolder, SOURCE_CONTROL_GIT_FOLDER); this.workflowExportFolder = path.join(this.gitFolder, SOURCE_CONTROL_WORKFLOW_EXPORT_FOLDER); @@ -1663,7 +1667,11 @@ export class SourceControlImportService { async deleteWorkflowsNotInWorkfolder(user: User, candidates: SourceControlledFile[]) { for (const candidate of candidates) { - await this.workflowService.delete(user, candidate.id, true); + try { + await this.workflowService.delete(user, candidate.id, true); + } catch (error) { + throw this.deletionError('workflow', [candidate], error); + } } } @@ -1701,9 +1709,27 @@ export class SourceControlImportService { } const candidateIds = candidates.map((c) => c.id); - await this.folderRepository.delete({ - id: In(candidateIds), - }); + try { + // Deleting a folder cascades to its subfolders and any workflows still + // inside the hierarchy, so those workflows must be prepared first. + const folderIds = [...candidateIds]; + for (const folderId of candidateIds) { + folderIds.push(...(await this.folderRepository.getAllFolderIdsInHierarchy(folderId))); + } + const workflows = await this.workflowRepository.find({ + select: ['id'], + where: { parentFolder: { id: In(folderIds) } }, + }); + await this.deactivateWorkflowsAndHardDeleteExecutions( + workflows.map((workflow) => workflow.id), + ); + + await this.folderRepository.delete({ + id: In(candidateIds), + }); + } catch (error) { + throw this.deletionError('folder', candidates, error); + } } async deleteTeamProjectsNotInWorkfolder(candidates: SourceControlledFile[]) { @@ -1712,9 +1738,81 @@ export class SourceControlImportService { } const candidateIds = candidates.map((c) => c.id); - await this.projectRepository.delete({ - id: In(candidateIds), - }); + try { + // Deleting a project cascades to its folders and workflows. Workflows are + // normally deleted individually before this point, but any still owned by + // the project (e.g. skipped for lack of permission) must be prepared for + // deletion via the cascade. + const ownedWorkflows = await this.sharedWorkflowRepository.find({ + select: ['workflowId'], + where: { projectId: In(candidateIds), role: 'workflow:owner' }, + }); + await this.deactivateWorkflowsAndHardDeleteExecutions( + ownedWorkflows.map((sw) => sw.workflowId), + ); + + await this.projectRepository.delete({ + id: In(candidateIds), + }); + } catch (error) { + throw this.deletionError('project', candidates, error); + } + } + + /** + * Deactivate the given workflows and hard-delete all their executions. + * To be called right before the workflows are removed via FK cascade + * (project/folder deletion): without it, the cascade can hit a DB statement + * timeout on large execution histories, and active workflows would keep + * their triggers registered in memory. + * + * Not using `WorkflowService.delete` here: it only deletes workflows the + * pulling user holds `workflow:delete` on, and the pull already ran it for + * those (see `deleteWorkflowsNotInWorkfolder`). Any workflow still standing + * was skipped by that permission check, yet the FK cascade below deletes it + * regardless — so we do the physical cleanup directly beforehand. Deletion + * hooks (`workflow.delete`/`workflow.afterDelete`) and the `workflow-deleted` + * event don't fire for these workflows — a pre-existing gap for any + * cascade-deleted workflow. + * + * REVIEW(question): should we instead extract the permission-free part of + * `WorkflowService.delete` (deactivate + drain + delete row + hooks/events) + * into an internal method and call it here? That would restore hook/event + * parity, at the cost of refactoring a hot service for this edge path. + */ + private async deactivateWorkflowsAndHardDeleteExecutions(workflowIds: string[]) { + for (const workflowId of workflowIds) { + const workflow = await this.workflowRepository.findOne({ + select: ['id', 'active'], + where: { id: workflowId }, + }); + if (!workflow) continue; + + if (workflow.active) { + await this.activeWorkflowManager.remove(workflow.id); + } + await this.executionPersistence.hardDeleteByWorkflowId(workflow.id); + } + } + + /** Contextual error for a failed deletion during pull, so the operator learns which resource to look at. */ + private deletionError( + type: 'workflow' | 'folder' | 'project', + candidates: SourceControlledFile[], + error: unknown, + ) { + const maxListed = 10; // keep the message readable when a pull deletes many resources + const resources = candidates + .slice(0, maxListed) + .map((c) => `"${c.name}" (${c.id})`) + .join(', '); + const overflow = + candidates.length > maxListed ? ` and ${candidates.length - maxListed} more` : ''; + const message = `Failed to delete ${type}(s) ${resources}${overflow} while pulling from source control: ${ensureError(error).message}`; + // report centrally: the controller rewraps this as a 4xx, which would + // otherwise hide an unexpected server-side failure from monitoring + this.errorReporter.error(ensureError(error), { extra: { message } }); + return new UnexpectedError(message, { cause: error }); } /** diff --git a/packages/cli/src/workflows/__tests__/workflow.service.test.ts b/packages/cli/src/workflows/__tests__/workflow.service.test.ts index adeb13ebaef..c0016df75f2 100644 --- a/packages/cli/src/workflows/__tests__/workflow.service.test.ts +++ b/packages/cli/src/workflows/__tests__/workflow.service.test.ts @@ -2,7 +2,6 @@ import type { Mock } from 'vitest'; import type { LicenseState } from '@n8n/backend-common'; import type { GlobalConfig, WorkflowsConfig } from '@n8n/config'; import type { - ExecutionRepository, Project, User, WorkflowRepository, @@ -17,6 +16,7 @@ import { mock } from 'vitest-mock-extended'; import type { IConnections, INode } from 'n8n-workflow'; import type { ActiveWorkflowManager } from '@/active-workflow-manager'; +import type { ExecutionPersistence } from '@/executions/execution-persistence'; import type { ScheduleTriggerJobRegistrar } from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar'; import { BadRequestError } from '@/errors/response-errors/bad-request.error'; import { ConflictError } from '@/errors/response-errors/conflict.error'; @@ -70,7 +70,6 @@ describe('WorkflowService', () => { mock(), // sharedWorkflowRepository workflowRepositoryMock as never, // workflowRepository mock(), // workflowTagMappingRepository - mock(), // binaryDataService mock(), // ownershipService mock(), // tagService mock(), // workflowHistoryService @@ -78,7 +77,7 @@ describe('WorkflowService', () => { mock(), // activeWorkflowManager roleServiceMock, // roleService mock(), // projectService - mock(), // executionRepository + mock(), // executionPersistence mock(), // eventService mock(), // globalConfig mock(), // folderRepository @@ -332,7 +331,6 @@ describe('WorkflowService', () => { mock(), // sharedWorkflowRepository workflowRepositoryMock as never, // workflowRepository mock(), // workflowTagMappingRepository - mock(), // binaryDataService ownershipServiceMock, // ownershipService mock(), // tagService workflowHistoryServiceMock, // workflowHistoryService @@ -340,7 +338,7 @@ describe('WorkflowService', () => { mock(), // activeWorkflowManager mock(), // roleService mock(), // projectService - mock(), // executionRepository + mock(), // executionPersistence mock(), // eventService mock(), // globalConfig mock(), // folderRepository @@ -1033,7 +1031,6 @@ describe('WorkflowService', () => { mock(), // sharedWorkflowRepository workflowRepositoryMock, // workflowRepository mock(), // workflowTagMappingRepository - mock(), // binaryDataService mock(), // ownershipService mock(), // tagService workflowHistoryServiceMock, // workflowHistoryService @@ -1041,7 +1038,7 @@ describe('WorkflowService', () => { activeWorkflowManagerMock, // activeWorkflowManager mock(), // roleService mock(), // projectService - mock(), // executionRepository + mock(), // executionPersistence eventServiceMock, // eventService globalConfigMock, // globalConfig mock(), // folderRepository @@ -1267,7 +1264,7 @@ describe('WorkflowService', () => { let workflowService: WorkflowService; let workflowFinderServiceMock: MockProxy; let workflowRepositoryMock: MockProxy; - let executionRepositoryMock: MockProxy; + let executionPersistenceMock: MockProxy; let globalConfigMock: MockProxy; let activeWorkflowManagerMock: MockProxy; @@ -1287,20 +1284,17 @@ describe('WorkflowService', () => { beforeEach(() => { workflowFinderServiceMock = mock(); workflowRepositoryMock = mock(); - executionRepositoryMock = mock(); + executionPersistenceMock = mock(); activeWorkflowManagerMock = mock(); globalConfigMock = mock({ workflows: mock({ useWorkflowPublicationService: true }), }); - executionRepositoryMock.find.mockResolvedValue([]); - workflowService = new WorkflowService( mock(), // logger mock(), // sharedWorkflowRepository workflowRepositoryMock, // workflowRepository mock(), // workflowTagMappingRepository - mock(), // binaryDataService mock(), // ownershipService mock(), // tagService mock(), // workflowHistoryService @@ -1308,7 +1302,7 @@ describe('WorkflowService', () => { activeWorkflowManagerMock, // activeWorkflowManager mock(), // roleService mock(), // projectService - executionRepositoryMock, // executionRepository + executionPersistenceMock, // executionPersistence mock(), // eventService globalConfigMock, // globalConfig mock(), // folderRepository @@ -1356,5 +1350,17 @@ describe('WorkflowService', () => { expect(workflowRepositoryMock.delete).toHaveBeenCalledWith(WORKFLOW_ID); }); + + test('deletes the workflow executions before the workflow itself', async () => { + const workflow = makeWorkflowEntity({ isArchived: true, activeVersionId: null }); + workflowFinderServiceMock.findWorkflowForUser.mockResolvedValue(workflow); + + await workflowService.delete(mock(), WORKFLOW_ID, true); + + expect(executionPersistenceMock.hardDeleteByWorkflowId).toHaveBeenCalledWith(WORKFLOW_ID); + expect( + executionPersistenceMock.hardDeleteByWorkflowId.mock.invocationCallOrder[0], + ).toBeLessThan(workflowRepositoryMock.delete.mock.invocationCallOrder[0]); + }); }); }); diff --git a/packages/cli/src/workflows/workflow.service.ts b/packages/cli/src/workflows/workflow.service.ts index 131bfa59b3c..c7f894ecc95 100644 --- a/packages/cli/src/workflows/workflow.service.ts +++ b/packages/cli/src/workflows/workflow.service.ts @@ -5,7 +5,6 @@ import type { User, ListQueryDb, WorkflowFolderUnionFull, WorkflowHistory } from import { SharedWorkflow, WorkflowEntity, - ExecutionRepository, FolderRepository, WorkflowTagMappingRepository, SharedWorkflowRepository, @@ -24,7 +23,6 @@ import { In } from '@n8n/typeorm'; import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity'; import isEqual from 'lodash/isEqual'; import pick from 'lodash/pick'; -import { FileLocation, BinaryDataService } from 'n8n-core'; import type { INode, INodes, IWorkflowSettings, JsonValue, IConnections } from 'n8n-workflow'; import { ensureError } from '@n8n/utils/errors/ensure-error'; import { PROJECT_ROOT, Workflow, assert, calculateWorkflowChecksum } from 'n8n-workflow'; @@ -40,6 +38,7 @@ import { WorkflowValidationError } from '@/errors/response-errors/workflow-valid import { WorkflowHistoryVersionNotFoundError } from '@/errors/workflow-history-version-not-found.error'; import { EventService } from '@/events/event.service'; import type { WorkflowActionSource } from '@/events/maps/relay.event-map'; +import { ExecutionPersistence } from '@/executions/execution-persistence'; import { ExternalHooks } from '@/external-hooks'; import { validateEntity } from '@/generic-helpers'; import { RedactionEnforcementService } from '@/modules/redaction/redaction-enforcement.service'; @@ -70,7 +69,6 @@ export class WorkflowService { private readonly sharedWorkflowRepository: SharedWorkflowRepository, private readonly workflowRepository: WorkflowRepository, private readonly workflowTagMappingRepository: WorkflowTagMappingRepository, - private readonly binaryDataService: BinaryDataService, private readonly ownershipService: OwnershipService, private readonly tagService: TagService, private readonly workflowHistoryService: WorkflowHistoryService, @@ -78,7 +76,7 @@ export class WorkflowService { private readonly activeWorkflowManager: ActiveWorkflowManager, private readonly roleService: RoleService, private readonly projectService: ProjectService, - private readonly executionRepository: ExecutionRepository, + private readonly executionPersistence: ExecutionPersistence, private readonly eventService: EventService, private readonly globalConfig: GlobalConfig, private readonly folderRepository: FolderRepository, @@ -1058,17 +1056,12 @@ export class WorkflowService { await this.activeWorkflowManager.remove(workflowId); } - const idsForDeletion = await this.executionRepository - .find({ - select: ['id'], - where: { workflowId }, - }) - .then((rows) => - rows.map(({ id: executionId }) => FileLocation.ofExecution(workflowId, executionId)), - ); + // Delete executions (incl. their binary and blob data) in batches before + // the workflow row, so the FK cascade on the workflow row stays too small + // to hit a DB statement timeout, however large the execution history. + await this.executionPersistence.hardDeleteByWorkflowId(workflowId); await this.workflowRepository.delete(workflowId); - await this.binaryDataService.deleteMany(idsForDeletion); this.eventService.emit('workflow-deleted', { user, workflowId, publicApi: false }); await this.externalHooks.run('workflow.afterDelete', [workflowId]); diff --git a/packages/cli/test/integration/environments/source-control-import.service.test.ts b/packages/cli/test/integration/environments/source-control-import.service.test.ts index 0fc6408d029..fcde5c2c267 100644 --- a/packages/cli/test/integration/environments/source-control-import.service.test.ts +++ b/packages/cli/test/integration/environments/source-control-import.service.test.ts @@ -124,6 +124,8 @@ describe('SourceControlImportService', () => { mock(), mock(), // redactionEnforcementService mock(), // dataTableSizeValidator + mock(), // activeWorkflowManager + mock(), // executionPersistence ); }); diff --git a/packages/cli/test/integration/workflows/workflow.service.test.ts b/packages/cli/test/integration/workflows/workflow.service.test.ts index 20077ba05d4..e8e7b34011e 100644 --- a/packages/cli/test/integration/workflows/workflow.service.test.ts +++ b/packages/cli/test/integration/workflows/workflow.service.test.ts @@ -69,7 +69,6 @@ beforeAll(async () => { Container.get(SharedWorkflowRepository), workflowRepository, mock(), - mock(), Container.get(OwnershipService), // ownershipService mock(), workflowHistoryService, @@ -77,7 +76,7 @@ beforeAll(async () => { activeWorkflowManager, Container.get(RoleService), // roleService Container.get(ProjectService), // projectService - mock(), // executionRepository + mock(), // executionPersistence mock(), // eventService globalConfig, mock(),