refactor(core): Route execution creation through ExecutionPersistence (#24912)

2026-11-04 23:35:45 +08:00 · 2026-01-30 12:46:53 +00:00
parent eb9978b890
commit c4ce0fbfbb
11 changed files with 287 additions and 128 deletions
@@ -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<ExecutionEntity> {
 		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<ExecutionEntity> {
 		} as IExecutionFlattedDb | IExecutionResponse | IExecutionBase;
 	}

-	/**
-	 * Insert a new execution and its execution data using a transaction.
-	 */
-	async createNewExecution(execution: CreateExecutionPayload): Promise<string> {
-		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];

@@ -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<ExecutionRepository>();
+const executionPersistence = mock<ExecutionPersistence>();

 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
@@ -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 });
@@ -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"},{}]');
+		});
+	});
+});
@@ -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<BinaryDataService>();
 	const fsStore = mock<FsStore>();

-	const executionPersistence = new ExecutionPersistence(
-		executionRepository,
-		binaryDataService,
-		fsStore,
-	);
-
 	beforeEach(() => {
 		jest.clearAllMocks();
 	});

+	const workflowData = mock<IWorkflowBase>({
+		id: 'workflow-123',
+		name: 'Test Workflow',
+		nodes: [],
+		connections: {},
+		versionId: 'version-abc',
+	});
+
+	const runData = createEmptyRunExecutionData();
+
+	const createMockTransaction = () => {
+		const mockTx = mock<EntityManager>();
+		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 <T>(cb: (em: EntityManager) => Promise<T>) => await cb(tx));
+
+	const createPersistenceService = (modeTag: 'db' | 'fs') =>
+		new ExecutionPersistence(
+			executionRepository,
+			binaryDataService,
+			fsStore,
+			mock<StorageConfig>({ 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<EntityManager>();
+				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<EntityManager>();
+				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 () => {
@@ -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;
 };

@@ -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;
@@ -366,7 +366,7 @@ export class ExecutionService {
 			status: 'error',
 		};

-		await this.executionRepository.createNewExecution(fullExecutionData);
+		await this.executionPersistence.create(fullExecutionData);
 	}

 	// ----------------------------------
@@ -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',
@@ -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,
@@ -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);