From c4ce0fbfbbf431ebcfb4433e694a3b6be4bdfbb5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iv=C3=A1n=20Ovejero?= Date: Fri, 30 Jan 2026 13:46:53 +0100 Subject: [PATCH] refactor(core): Route execution creation through `ExecutionPersistence` (#24912) --- .../src/repositories/execution.repository.ts | 26 --- .../src/__tests__/active-executions.test.ts | 12 +- packages/cli/src/active-executions.ts | 4 +- .../execution-persistence.integration.test.ts | 59 +++++++ .../__tests__/execution-persistence.test.ts | 151 +++++++++++++++++- .../src/executions/execution-data/types.ts | 7 +- .../src/executions/execution-persistence.ts | 50 +++++- .../cli/src/executions/execution.service.ts | 2 +- .../chat-hub.service.integration.test.ts | 31 ++-- .../workflows/workflow-execution.service.ts | 7 +- .../repositories/execution.repository.test.ts | 66 -------- 11 files changed, 287 insertions(+), 128 deletions(-) create mode 100644 packages/cli/src/executions/__tests__/execution-persistence.integration.test.ts diff --git a/packages/@n8n/db/src/repositories/execution.repository.ts b/packages/@n8n/db/src/repositories/execution.repository.ts index 2f779c22aa4..94e80d478f0 100644 --- a/packages/@n8n/db/src/repositories/execution.repository.ts +++ b/packages/@n8n/db/src/repositories/execution.repository.ts @@ -39,7 +39,6 @@ import { } from 'n8n-workflow'; import * as a from 'node:assert/strict'; -import { ExecutionDataRepository } from './execution-data.repository'; import { AnnotationTagEntity, AnnotationTagMapping, @@ -52,7 +51,6 @@ import { WorkflowEntity, } from '../entities'; import type { - CreateExecutionPayload, ExecutionSummaries, IExecutionBase, IExecutionFlattedDb, @@ -154,7 +152,6 @@ export class ExecutionRepository extends Repository { private readonly globalConfig: GlobalConfig, private readonly logger: Logger, private readonly errorReporter: ErrorReporter, - private readonly executionDataRepository: ExecutionDataRepository, private readonly binaryDataService: BinaryDataService, ) { super(ExecutionEntity, dataSource.manager); @@ -342,29 +339,6 @@ export class ExecutionRepository extends Repository { } as IExecutionFlattedDb | IExecutionResponse | IExecutionBase; } - /** - * Insert a new execution and its execution data using a transaction. - */ - async createNewExecution(execution: CreateExecutionPayload): Promise { - const { data: dataObj, workflowData: currentWorkflow, ...rest } = execution; - const { connections, nodes, name, settings } = currentWorkflow ?? {}; - const workflowData = { connections, nodes, name, settings, id: currentWorkflow.id }; - const data = stringify(dataObj); - - return await this.manager.transaction(async (transactionManager) => { - const { identifiers: inserted } = await transactionManager.insert(ExecutionEntity, { - ...rest, - createdAt: new Date(), - }); - const { id: executionId } = inserted[0] as { id: string }; - await this.executionDataRepository.createExecutionDataForExecution( - { executionId, workflowData, data, workflowVersionId: currentWorkflow.versionId }, - transactionManager, - ); - return String(executionId); - }); - } - async markAsCrashed(executionIds: string | string[]) { if (!Array.isArray(executionIds)) executionIds = [executionIds]; diff --git a/packages/cli/src/__tests__/active-executions.test.ts b/packages/cli/src/__tests__/active-executions.test.ts index cfc0247eb07..9438b70048e 100644 --- a/packages/cli/src/__tests__/active-executions.test.ts +++ b/packages/cli/src/__tests__/active-executions.test.ts @@ -22,6 +22,7 @@ import { v4 as uuid } from 'uuid'; import { ActiveExecutions } from '@/active-executions'; import { ConcurrencyControlService } from '@/concurrency/concurrency-control.service'; +import type { ExecutionPersistence } from '@/executions/execution-persistence'; jest.mock('n8n-workflow', () => ({ ...jest.requireActual('n8n-workflow'), @@ -32,6 +33,7 @@ const FAKE_EXECUTION_ID = '15'; const FAKE_SECOND_EXECUTION_ID = '20'; const executionRepository = mock(); +const executionPersistence = mock(); const concurrencyControl = mockInstance(ConcurrencyControlService, { // @ts-expect-error Private property @@ -76,12 +78,13 @@ describe('ActiveExecutions', () => { activeExecutions = new ActiveExecutions( mock(), executionRepository, + executionPersistence, concurrencyControl, mock(), executionsConfig, ); - executionRepository.createNewExecution.mockResolvedValue(FAKE_EXECUTION_ID); + executionPersistence.create.mockResolvedValue(FAKE_EXECUTION_ID); executionRepository.updateExistingExecution.mockResolvedValue(true); executionRepository.setRunning.mockResolvedValue(Promise.resolve(new Date())); @@ -103,7 +106,7 @@ describe('ActiveExecutions', () => { expect(executionId).toBe(FAKE_EXECUTION_ID); expect(activeExecutions.getActiveExecutions()).toHaveLength(1); - expect(executionRepository.createNewExecution).toHaveBeenCalledTimes(1); + expect(executionPersistence.create).toHaveBeenCalledTimes(1); expect(executionRepository.updateExistingExecution).toHaveBeenCalledTimes(0); }); @@ -112,7 +115,7 @@ describe('ActiveExecutions', () => { expect(executionId).toBe(FAKE_SECOND_EXECUTION_ID); expect(activeExecutions.getActiveExecutions()).toHaveLength(1); - expect(executionRepository.createNewExecution).toHaveBeenCalledTimes(0); + expect(executionPersistence.create).toHaveBeenCalledTimes(0); expect(executionRepository.updateExistingExecution).toHaveBeenCalledTimes(1); }); @@ -259,6 +262,7 @@ describe('ActiveExecutions', () => { activeExecutions = new ActiveExecutions( logger, executionRepository, + executionPersistence, concurrencyControl, mock(), executionsConfig, @@ -361,7 +365,7 @@ describe('ActiveExecutions', () => { beforeEach(async () => { let i = 1000; - executionRepository.createNewExecution.mockImplementation(async () => `${i++}`); + executionPersistence.create.mockImplementation(async () => `${i++}`); (sleep as jest.Mock).mockImplementation(() => { // @ts-expect-error private property diff --git a/packages/cli/src/active-executions.ts b/packages/cli/src/active-executions.ts index 5af05909802..d7ff38a609a 100644 --- a/packages/cli/src/active-executions.ts +++ b/packages/cli/src/active-executions.ts @@ -23,6 +23,7 @@ import type PCancelable from 'p-cancelable'; import { ExecutionNotFoundError } from '@/errors/execution-not-found-error'; import { ExecutionAlreadyResumingError } from '@/errors/execution-already-resuming.error'; +import { ExecutionPersistence } from '@/executions/execution-persistence'; import type { IExecutingWorkflowData, IExecutionsCurrentSummary } from '@/interfaces'; import { isWorkflowIdValid } from '@/utils'; @@ -45,6 +46,7 @@ export class ActiveExecutions { constructor( private readonly logger: Logger, private readonly executionRepository: ExecutionRepository, + private readonly executionPersistence: ExecutionPersistence, private readonly concurrencyControl: ConcurrencyControlService, private readonly eventService: EventService, private readonly executionsConfig: ExecutionsConfig, @@ -82,7 +84,7 @@ export class ActiveExecutions { fullExecutionData.workflowId = workflowId; } - maybeExecutionId = await this.executionRepository.createNewExecution(fullExecutionData); + maybeExecutionId = await this.executionPersistence.create(fullExecutionData); assert(maybeExecutionId); await capacityReservation.reserve({ mode, executionId: maybeExecutionId }); diff --git a/packages/cli/src/executions/__tests__/execution-persistence.integration.test.ts b/packages/cli/src/executions/__tests__/execution-persistence.integration.test.ts new file mode 100644 index 00000000000..47678ca00cf --- /dev/null +++ b/packages/cli/src/executions/__tests__/execution-persistence.integration.test.ts @@ -0,0 +1,59 @@ +import { createWorkflow, testDb } from '@n8n/backend-test-utils'; +import { ExecutionDataRepository, ExecutionRepository } from '@n8n/db'; +import { Container } from '@n8n/di'; + +import { ExecutionPersistence } from '@/executions/execution-persistence'; + +describe('ExecutionPersistence', () => { + beforeAll(async () => { + await testDb.init(); + }); + + beforeEach(async () => { + await testDb.truncate(['WorkflowEntity', 'ExecutionEntity']); + }); + + afterAll(async () => { + await testDb.terminate(); + }); + + describe('create', () => { + it('should save execution and execution data to database', async () => { + const executionRepo = Container.get(ExecutionRepository); + const executionPersistence = Container.get(ExecutionPersistence); + const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } }); + + const executionId = await executionPersistence.create({ + workflowId: workflow.id, + data: { + // @ts-expect-error Partial data for test + resultData: {}, + }, + workflowData: workflow, + mode: 'manual', + startedAt: new Date(), + status: 'new', + finished: false, + }); + + expect(executionId).toBeDefined(); + + const executionEntity = await executionRepo.findOneBy({ id: executionId }); + expect(executionEntity?.id).toEqual(executionId); + expect(executionEntity?.workflowId).toEqual(workflow.id); + expect(executionEntity?.status).toEqual('new'); + expect(executionEntity?.storedAt).toEqual('db'); + + const executionDataRepository = Container.get(ExecutionDataRepository); + const executionData = await executionDataRepository.findOneBy({ executionId }); + expect(executionData?.workflowData).toEqual({ + id: workflow.id, + connections: workflow.connections, + nodes: workflow.nodes, + name: workflow.name, + settings: workflow.settings, + }); + expect(executionData?.data).toEqual('[{"resultData":"1"},{}]'); + }); + }); +}); diff --git a/packages/cli/src/executions/__tests__/execution-persistence.test.ts b/packages/cli/src/executions/__tests__/execution-persistence.test.ts index 52165e2c622..6a7a56e7b01 100644 --- a/packages/cli/src/executions/__tests__/execution-persistence.test.ts +++ b/packages/cli/src/executions/__tests__/execution-persistence.test.ts @@ -1,8 +1,17 @@ +/* eslint-disable id-denylist */ /* eslint-disable @typescript-eslint/unbound-method */ -import type { ExecutionRepository } from '@n8n/db'; +import { + ExecutionData, + ExecutionEntity, + type CreateExecutionPayload, + type EntityManager, + type ExecutionRepository, +} from '@n8n/db'; import { mock } from 'jest-mock-extended'; -import type { BinaryDataService } from 'n8n-core'; +import type { BinaryDataService, StorageConfig } from 'n8n-core'; +import type { IWorkflowBase } from 'n8n-workflow'; +import { createEmptyRunExecutionData } from 'n8n-workflow'; import type { FsStore } from '@/executions/execution-data/fs-store'; import { ExecutionPersistence } from '@/executions/execution-persistence'; @@ -12,17 +21,143 @@ describe('ExecutionPersistence', () => { const binaryDataService = mock(); const fsStore = mock(); - const executionPersistence = new ExecutionPersistence( - executionRepository, - binaryDataService, - fsStore, - ); - beforeEach(() => { jest.clearAllMocks(); }); + const workflowData = mock({ + id: 'workflow-123', + name: 'Test Workflow', + nodes: [], + connections: {}, + versionId: 'version-abc', + }); + + const runData = createEmptyRunExecutionData(); + + const createMockTransaction = () => { + const mockTx = mock(); + mockTx.insert.mockResolvedValue({ + identifiers: [{ id: 'exec-1' }], + generatedMaps: [], + raw: {}, + }); + mockTx.update.mockResolvedValue({ affected: 1, generatedMaps: [], raw: {} }); + return mockTx; + }; + + const createMockTx = (tx: EntityManager) => + jest.fn().mockImplementation(async (cb: (em: EntityManager) => Promise) => await cb(tx)); + + const createPersistenceService = (modeTag: 'db' | 'fs') => + new ExecutionPersistence( + executionRepository, + binaryDataService, + fsStore, + mock({ modeTag }), + ); + + describe('create', () => { + const createPayload: CreateExecutionPayload = { + data: runData, + workflowData, + mode: 'manual', + finished: false, + status: 'new', + workflowId: 'workflow-123', + }; + + describe('database mode', () => { + const executionPersistence = createPersistenceService('db'); + + it('should create execution with `storedAt: db` and insert data via transaction', async () => { + const mockTx = createMockTransaction(); + executionRepository.manager.transaction = createMockTx(mockTx); + + const executionId = await executionPersistence.create(createPayload); + + expect(executionId).toBe('exec-1'); + expect(mockTx.insert).toHaveBeenCalledWith( + ExecutionEntity, + expect.objectContaining({ + storedAt: 'db', + mode: 'manual', + finished: false, + status: 'new', + workflowId: 'workflow-123', + createdAt: expect.any(Date) as Date, + }), + ); + expect(mockTx.insert).toHaveBeenCalledWith( + ExecutionData, + expect.objectContaining({ + executionId: 'exec-1', + workflowVersionId: 'version-abc', + }), + ); + expect(mockTx.insert).toHaveBeenCalledTimes(2); + expect(fsStore.write).not.toHaveBeenCalled(); + }); + }); + + describe('filesystem mode', () => { + const executionPersistence = createPersistenceService('fs'); + + it('should create execution with `storedAt: fs` and write data to filesystem', async () => { + const mockTx = mock(); + mockTx.insert.mockResolvedValue({ + identifiers: [{ id: 'exec-2' }], + generatedMaps: [], + raw: {}, + }); + executionRepository.manager.transaction = createMockTx(mockTx); + + const executionId = await executionPersistence.create(createPayload); + + expect(executionId).toBe('exec-2'); + expect(mockTx.insert).toHaveBeenCalledWith( + ExecutionEntity, + expect.objectContaining({ + storedAt: 'fs', + }), + ); + expect(mockTx.insert).toHaveBeenCalledTimes(1); + const expectedWorkflowSnapshot = { + id: workflowData.id, + name: workflowData.name, + nodes: workflowData.nodes, + connections: workflowData.connections, + settings: workflowData.settings, + }; + expect(fsStore.write).toHaveBeenCalledWith( + { workflowId: 'workflow-123', executionId: 'exec-2' }, + expect.objectContaining({ + workflowData: expectedWorkflowSnapshot, + workflowVersionId: 'version-abc', + }), + ); + }); + + it('should roll back transaction if filesystem write fails', async () => { + const mockTx = mock(); + mockTx.insert.mockResolvedValue({ + identifiers: [{ id: 'exec-3' }], + generatedMaps: [], + raw: {}, + }); + + const fsWriteError = new Error('Filesystem write failed'); + fsStore.write.mockRejectedValue(fsWriteError); + + executionRepository.manager.transaction = createMockTx(mockTx); + + await expect(executionPersistence.create(createPayload)).rejects.toThrow(fsWriteError); + }); + }); + }); + describe('hardDelete', () => { + const executionPersistence = createPersistenceService('db'); const baseTarget = { workflowId: 'wf-1', executionId: 'exec-1' }; it('should delete execution, binary data, and fs data when storedAt is fs', async () => { diff --git a/packages/cli/src/executions/execution-data/types.ts b/packages/cli/src/executions/execution-data/types.ts index 161186ded1c..67b60798b0b 100644 --- a/packages/cli/src/executions/execution-data/types.ts +++ b/packages/cli/src/executions/execution-data/types.ts @@ -9,9 +9,14 @@ export function createExecutionRef(workflowId: string, executionId: string): Exe return { workflowId, executionId }; } +export type WorkflowSnapshot = Pick< + IWorkflowBase, + 'id' | 'name' | 'nodes' | 'connections' | 'settings' +>; + export type ExecutionDataPayload = { data: string; - workflowData: IWorkflowBase; + workflowData: WorkflowSnapshot; workflowVersionId: string | null; }; diff --git a/packages/cli/src/executions/execution-persistence.ts b/packages/cli/src/executions/execution-persistence.ts index 32237820154..99f002c6761 100644 --- a/packages/cli/src/executions/execution-persistence.ts +++ b/packages/cli/src/executions/execution-persistence.ts @@ -1,10 +1,15 @@ -import type { ExecutionDataStorageLocation, ExecutionDeletionCriteria } from '@n8n/db'; -import { ExecutionRepository } from '@n8n/db'; +import type { + CreateExecutionPayload, + ExecutionDataStorageLocation, + ExecutionDeletionCriteria, +} from '@n8n/db'; +import { ExecutionData, ExecutionEntity, ExecutionRepository } from '@n8n/db'; import { Service } from '@n8n/di'; -import { BinaryDataService } from 'n8n-core'; +import { stringify } from 'flatted'; +import { BinaryDataService, StorageConfig } from 'n8n-core'; import { FsStore } from './execution-data/fs-store'; -import type { ExecutionRef } from './execution-data/types'; +import type { ExecutionRef, WorkflowSnapshot } from './execution-data/types'; type DeletionTarget = ExecutionRef & { storedAt: ExecutionDataStorageLocation }; @@ -18,8 +23,45 @@ export class ExecutionPersistence { private readonly executionRepository: ExecutionRepository, private readonly binaryDataService: BinaryDataService, private readonly fsStore: FsStore, + private readonly storageConfig: StorageConfig, ) {} + /** + * Create an execution entity and persist its data to the configured storage. + * - In `db` mode, we write both entity and data to the DB in a transaction. + * - In `fs` mode, we write the entity to the DB and its data to the filesystem. + */ + async create(payload: CreateExecutionPayload) { + const { data: rawData, workflowData, ...rest } = payload; + const { connections, nodes, name, settings, id } = workflowData; + const workflowSnapshot: WorkflowSnapshot = { connections, nodes, name, settings, id }; + const storedAt = this.storageConfig.modeTag; + const executionEntity = { ...rest, createdAt: new Date(), storedAt }; + const data = stringify(rawData); + const workflowVersionId = workflowData.versionId ?? null; + + return await this.executionRepository.manager.transaction(async (tx) => { + const { identifiers } = await tx.insert(ExecutionEntity, executionEntity); + const executionId = String(identifiers[0].id); + + if (storedAt === 'db') { + await tx.insert(ExecutionData, { + executionId, + workflowData: workflowSnapshot, + data, + workflowVersionId, + }); + return executionId; + } + + await this.fsStore.write( + { workflowId: id, executionId }, + { data, workflowData: workflowSnapshot, workflowVersionId }, + ); + return executionId; + }); + } + async hardDelete(target: DeletionTarget | DeletionTarget[]) { const targets = Array.isArray(target) ? target : [target]; if (targets.length === 0) return; diff --git a/packages/cli/src/executions/execution.service.ts b/packages/cli/src/executions/execution.service.ts index c5abcc95efd..9385f450d86 100644 --- a/packages/cli/src/executions/execution.service.ts +++ b/packages/cli/src/executions/execution.service.ts @@ -366,7 +366,7 @@ export class ExecutionService { status: 'error', }; - await this.executionRepository.createNewExecution(fullExecutionData); + await this.executionPersistence.create(fullExecutionData); } // ---------------------------------- diff --git a/packages/cli/src/modules/chat-hub/__tests__/chat-hub.service.integration.test.ts b/packages/cli/src/modules/chat-hub/__tests__/chat-hub.service.integration.test.ts index 80fe8abcc37..1d0987f941b 100644 --- a/packages/cli/src/modules/chat-hub/__tests__/chat-hub.service.integration.test.ts +++ b/packages/cli/src/modules/chat-hub/__tests__/chat-hub.service.integration.test.ts @@ -21,6 +21,7 @@ import { createAdmin, createMember } from '@test-integration/db/users'; import { retryUntil } from '@test-integration/retry-until'; import { ActiveExecutions } from '../../../active-executions'; +import { ExecutionPersistence } from '../../../executions/execution-persistence'; import { ChatExecutionManager } from '../../../chat/chat-execution-manager'; import { WorkflowExecutionService } from '../../../workflows/workflow-execution.service'; import { ChatHubAgentRepository } from '../chat-hub-agent.repository'; @@ -62,6 +63,7 @@ describe('chatHub', () => { let sessionsRepository: ChatHubSessionRepository; let agentRepository: ChatHubAgentRepository; let executionRepository: ExecutionRepository; + let executionPersistence: ExecutionPersistence; let instanceSettings: InstanceSettings; let settingsRepository: SettingsRepository; @@ -74,6 +76,7 @@ describe('chatHub', () => { sessionsRepository = Container.get(ChatHubSessionRepository); agentRepository = Container.get(ChatHubAgentRepository); executionRepository = Container.get(ExecutionRepository); + executionPersistence = Container.get(ExecutionPersistence); instanceSettings = Container.get(InstanceSettings); settingsRepository = Container.get(SettingsRepository); }); @@ -959,7 +962,7 @@ describe('chatHub', () => { it('should respond and persist generated response chunks sent from workflow execution', async () => { // First call: main message execution with stream spyExecute.mockImplementationOnce(async (_user, workflowData, executionData, stream) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'chat', status: 'running', @@ -1046,7 +1049,7 @@ describe('chatHub', () => { it('should respond and persist an error chunk sent from workflow execution', async () => { // First call: main message execution with stream spyExecute.mockImplementationOnce(async (_user, workflowData, executionData, stream) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'chat', status: 'running', @@ -1128,7 +1131,7 @@ describe('chatHub', () => { it('should respond and persist an error set in the workflow execution', async () => { // First call: main message execution with stream spyExecute.mockImplementationOnce(async (_user, workflowData, executionData, stream) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'chat', status: 'running', @@ -1244,7 +1247,7 @@ describe('chatHub', () => { // First call: main message execution with stream that closes quickly spyExecute.mockImplementationOnce(async (_u, workflowData, data, stream) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'chat', status: 'running', @@ -1330,7 +1333,7 @@ describe('chatHub', () => { // First call: main message execution with stream that never closes spyExecute.mockImplementationOnce(async (_u, workflowData, data, stream) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'chat', status: 'running', @@ -1497,7 +1500,7 @@ describe('chatHub', () => { // Mock the execution to return lastNode output spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', @@ -1635,7 +1638,7 @@ describe('chatHub', () => { // Mock the execution to return an error spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', @@ -1743,7 +1746,7 @@ describe('chatHub', () => { // Mock the execution to return text field instead of output spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', @@ -1865,7 +1868,7 @@ describe('chatHub', () => { // Mock initial execution - returns waiting status spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', @@ -2024,7 +2027,7 @@ describe('chatHub', () => { // Mock execution - returns waiting status spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', @@ -2221,7 +2224,7 @@ describe('chatHub', () => { ); spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', @@ -2349,7 +2352,7 @@ describe('chatHub', () => { ); spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', @@ -2475,7 +2478,7 @@ describe('chatHub', () => { ); spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', @@ -2572,7 +2575,7 @@ describe('chatHub', () => { ); spyExecute.mockImplementationOnce(async (_user, workflowData, executionData) => { - const executionId = await executionRepository.createNewExecution({ + const executionId = await executionPersistence.create({ finished: false, mode: 'webhook', status: 'running', diff --git a/packages/cli/src/workflows/workflow-execution.service.ts b/packages/cli/src/workflows/workflow-execution.service.ts index ef83560f865..88ec67b87c3 100644 --- a/packages/cli/src/workflows/workflow-execution.service.ts +++ b/packages/cli/src/workflows/workflow-execution.service.ts @@ -1,7 +1,7 @@ import { Logger } from '@n8n/backend-common'; import { GlobalConfig } from '@n8n/config'; import type { Project, User, CreateExecutionPayload } from '@n8n/db'; -import { ExecutionRepository, WorkflowRepository } from '@n8n/db'; +import { WorkflowRepository } from '@n8n/db'; import { Service } from '@n8n/di'; import type { Response } from 'express'; import { DirectedGraph, ErrorReporter, anyReachableRootHasRunData } from 'n8n-core'; @@ -26,6 +26,7 @@ import { } from 'n8n-workflow'; import { EventService } from '@/events/event.service'; +import { ExecutionPersistence } from '@/executions/execution-persistence'; import { FailedRunFactory } from '@/executions/failed-run-factory'; import { SubworkflowPolicyChecker } from '@/executions/pre-execution-checks'; import type { IWorkflowErrorData } from '@/interfaces'; @@ -40,7 +41,7 @@ export class WorkflowExecutionService { constructor( private readonly logger: Logger, private readonly errorReporter: ErrorReporter, - private readonly executionRepository: ExecutionRepository, + private readonly executionPersistence: ExecutionPersistence, private readonly workflowRepository: WorkflowRepository, private readonly nodeTypes: NodeTypes, private readonly testWebhooks: TestWebhooks, @@ -359,7 +360,7 @@ export class WorkflowExecutionService { workflowId: workflowData.id, }; - await this.executionRepository.createNewExecution(fullExecutionData); + await this.executionPersistence.create(fullExecutionData); } this.logger.info('Error workflow execution blocked due to subworkflow settings', { erroredWorkflowId: workflowErrorData.workflow.id, diff --git a/packages/cli/test/integration/database/repositories/execution.repository.test.ts b/packages/cli/test/integration/database/repositories/execution.repository.test.ts index 3af2af14145..a9327a15245 100644 --- a/packages/cli/test/integration/database/repositories/execution.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/execution.repository.test.ts @@ -17,72 +17,6 @@ describe('ExecutionRepository', () => { await testDb.terminate(); }); - describe('createNewExecution', () => { - it('should save execution data', async () => { - const executionRepo = Container.get(ExecutionRepository); - const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } }); - const executionId = await executionRepo.createNewExecution({ - workflowId: workflow.id, - data: { - //@ts-expect-error This is not needed for tests - resultData: {}, - }, - workflowData: workflow, - mode: 'manual', - startedAt: new Date(), - status: 'new', - finished: false, - }); - - expect(executionId).toBeDefined(); - - const executionEntity = await executionRepo.findOneBy({ id: executionId }); - expect(executionEntity?.id).toEqual(executionId); - expect(executionEntity?.workflowId).toEqual(workflow.id); - expect(executionEntity?.status).toEqual('new'); - - const executionDataRepo = Container.get(ExecutionDataRepository); - const executionData = await executionDataRepo.findOneBy({ executionId }); - expect(executionData?.workflowData).toEqual({ - id: workflow.id, - connections: workflow.connections, - nodes: workflow.nodes, - name: workflow.name, - settings: workflow.settings, - }); - expect(executionData?.data).toEqual('[{"resultData":"1"},{}]'); - }); - - it('should not create execution if execution data insert fails', async () => { - const executionRepo = Container.get(ExecutionRepository); - const executionDataRepo = Container.get(ExecutionDataRepository); - - const workflow = await createWorkflow({ settings: { executionOrder: 'v1' } }); - jest - .spyOn(executionDataRepo, 'createExecutionDataForExecution') - .mockRejectedValueOnce(new Error()); - - await expect( - async () => - await executionRepo.createNewExecution({ - workflowId: workflow.id, - data: { - //@ts-expect-error This is not needed for tests - resultData: {}, - }, - workflowData: workflow, - mode: 'manual', - startedAt: new Date(), - status: 'new', - finished: false, - }), - ).rejects.toThrow(); - - const executionEntities = await executionRepo.find(); - expect(executionEntities).toBeEmptyArray(); - }); - }); - describe('run execution data migration', () => { it('should automatically migrate IRunExecutionDataV0 to V1 when reading', async () => { const executionRepo = Container.get(ExecutionRepository);