feat(core): Add Prometheus metrics for execution data (#31655)

packages/@n8n/config/src/configs/endpoints.config.ts

@@ -79,6 +79,10 @@ class PrometheusMetricsConfig {
 	/** How often (in seconds) to update workflow statistics metrics. */
 	@Env('N8N_METRICS_WORKFLOW_STATISTICS_INTERVAL')
 	workflowStatisticsInterval: number = 300;
+
+	/** Whether to include metrics for execution data reads and writes. */
+	@Env('N8N_METRICS_INCLUDE_EXECUTION_DATA_METRICS')
+	includeExecutionDataMetrics: boolean = false;
 }
 
 @Config

packages/@n8n/config/test/config.test.ts

@@ -235,6 +235,7 @@ describe('GlobalConfig', () => {
 				activeWorkflowCountInterval: 60,
 				includeWorkflowStatistics: false,
 				workflowStatisticsInterval: 300,
+				includeExecutionDataMetrics: false,
 			},
 			additionalNonUIRoutes: '',
 			disableProductionWebhooksOnMainProcess: false,

packages/cli/src/events/__tests__/telemetry-event-relay.test.ts

@@ -62,6 +62,7 @@ describe('TelemetryEventRelay', () => {
 				includeCacheMetrics: false,
 				includeMessageEventBusMetrics: false,
 				includeQueueMetrics: false,
+				includeExecutionDataMetrics: false,
 			},
 		},
 		logging: {
@@ -2105,6 +2106,7 @@ describe('TelemetryEventRelay', () => {
 						metrics_category_logs: false,
 						metrics_category_queue: false,
 						metrics_category_routes: false,
+						metrics_category_execution_data: false,
 						metrics_enabled: true,
 					},
 					n8n_binary_data_mode: 'default',
@@ -2142,6 +2144,7 @@ describe('TelemetryEventRelay', () => {
 						metrics_category_cache: false,
 						metrics_category_logs: false,
 						metrics_category_queue: false,
+						metrics_category_execution_data: false,
 					},
 					otel: {
 						enabled: false,

packages/cli/src/events/event.service.ts

@@ -3,10 +3,11 @@ import { Service } from '@n8n/di';
 import { TypedEmitter } from '@/typed-emitter';
 
 import type { AiEventMap } from './maps/ai.event-map';
+import type { ExecutionDataEventMap } from './maps/execution-data.event-map';
 import type { QueueMetricsEventMap } from './maps/queue-metrics.event-map';
 import type { RelayEventMap } from './maps/relay.event-map';
 
-type EventMap = RelayEventMap & QueueMetricsEventMap & AiEventMap;
+type EventMap = RelayEventMap & QueueMetricsEventMap & AiEventMap & ExecutionDataEventMap;
 
 @Service()
 export class EventService extends TypedEmitter<EventMap> {}

packages/cli/src/events/maps/execution-data.event-map.ts

@@ -0,0 +1,25 @@
+import type { ExecutionDataStorageLocation } from '@n8n/db';
+
+/**
+ * A "read" spans fetching the bundle from its store AND deserializing it into usable data, so a
+ * deserialization failure counts as a failed read and its duration includes deserialize time.
+ * A "write" is symmetric: it spans serializing the data AND writing it to its store.
+ *
+ * `unreadableBundles` counts bundles that were missing (the store had none) or corrupt (fetched
+ * but failed to deserialize) — the data-loss signal. It excludes infrastructure failures (e.g. the
+ * store being unreachable), where the bundles' integrity is unknown; those are failed reads with
+ * `unreadableBundles: 0`.
+ */
+export type ExecutionDataEventMap = {
+	'execution-data-read': {
+		mode: ExecutionDataStorageLocation;
+		durationMs: number;
+		success: boolean;
+		unreadableBundles: number;
+	};
+	'execution-data-write': {
+		mode: ExecutionDataStorageLocation;
+		durationMs: number;
+		success: boolean;
+	};
+};

packages/cli/src/events/relays/telemetry.event-relay.ts

@@ -1296,6 +1296,8 @@ export class TelemetryEventRelay extends EventRelay {
 				metrics_category_cache: this.globalConfig.endpoints.metrics.includeCacheMetrics,
 				metrics_category_logs: this.globalConfig.endpoints.metrics.includeMessageEventBusMetrics,
 				metrics_category_queue: this.globalConfig.endpoints.metrics.includeQueueMetrics,
+				metrics_category_execution_data:
+					this.globalConfig.endpoints.metrics.includeExecutionDataMetrics,
 			},
 		};
 

packages/cli/src/executions/__tests__/execution-persistence.test.ts

@@ -16,8 +16,10 @@ import type { IWorkflowBase } from 'n8n-workflow';
 import { createEmptyRunExecutionData, UnexpectedError } from 'n8n-workflow';
 
 import { DuplicateExecutionError } from '@/errors/duplicate-execution.error';
+import type { EventService } from '@/events/event.service';
 import type { DbStore } from '@/executions/execution-data/db-store';
 import type { FsStore } from '@/executions/execution-data/fs-store';
+import { CorruptedExecutionDataError } from '@/executions/execution-data/corrupted-execution-data.error';
 import { MissingExecutionDataError } from '@/executions/execution-data/missing-execution-data.error';
 import { ExecutionPersistence } from '@/executions/execution-persistence';
 
@@ -27,6 +29,7 @@ describe('ExecutionPersistence', () => {
 	const fsStore = mock<FsStore>();
 	const dbStore = mock<DbStore>();
 	const errorReporter = mock<ErrorReporter>();
+	const eventService = mock<EventService>();
 	const executionsConfig = mock<ExecutionsConfig>({
 		pruneData: true,
 		pruneDataHardDeleteBuffer: 1,
@@ -73,6 +76,7 @@ describe('ExecutionPersistence', () => {
 			executionsConfig,
 			mock<DatabaseConfig>({ type: dbType }),
 			errorReporter,
+			eventService,
 		);
 
 	describe('create', () => {
@@ -1146,7 +1150,8 @@ describe('ExecutionPersistence', () => {
 
 			const result = await executionPersistence.findMultipleExecutions({}, { includeData: true });
 
-			expect(executionRepository.reportInvalidExecutions).toHaveBeenCalledWith([dbB, fsC]);
+			expect(executionRepository.reportInvalidExecutions).toHaveBeenCalledWith([dbB]);
+			expect(executionRepository.reportInvalidExecutions).toHaveBeenCalledWith([fsC]);
 			expect(result.map((e) => e.id)).toEqual(['a']);
 		});
 
@@ -1356,4 +1361,277 @@ describe('ExecutionPersistence', () => {
 			expect(executionRepository.update).not.toHaveBeenCalled();
 		});
 	});
+
+	describe('execution data metrics events', () => {
+		const createPayload: CreateExecutionPayload = {
+			data: runData,
+			workflowData,
+			mode: 'manual',
+			finished: false,
+			status: 'new',
+			workflowId: 'workflow-123',
+		};
+
+		const readBundle = {
+			data: '[{"resultData":"1"},{}]',
+			workflowData: { id: 'wf-1', name: 's', nodes: [], connections: {}, settings: undefined },
+			workflowVersionId: 'v-1',
+			version: 1 as const,
+		};
+
+		const entity = (storedAt: 'db' | 'fs', id = 'exec-1') =>
+			({
+				id,
+				workflowId: 'wf-1',
+				storedAt,
+				metadata: [],
+				annotation: undefined,
+				status: 'success',
+			}) as unknown as ExecutionEntity;
+
+		it('emits a successful write event on create', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.manager.transaction = createMockTx(createMockTransaction());
+
+			await executionPersistence.create(createPayload);
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-write',
+				expect.objectContaining({ mode: 'db', success: true, durationMs: expect.any(Number) }),
+			);
+		});
+
+		it('emits a failed write event when the store write throws', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.manager.transaction = createMockTx(createMockTransaction());
+			dbStore.write.mockRejectedValueOnce(new Error('disk full'));
+
+			await expect(executionPersistence.create(createPayload)).rejects.toThrow('disk full');
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-write',
+				expect.objectContaining({ mode: 'db', success: false }),
+			);
+		});
+
+		it('emits a successful read event when a single bundle is found', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.findOne.mockResolvedValue(entity('db'));
+			dbStore.read.mockResolvedValue(readBundle);
+
+			await executionPersistence.findSingleExecution('exec-1', { includeData: true });
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: true, unreadableBundles: 0 }),
+			);
+		});
+
+		it('emits a failed read with one unreadable bundle when a single bundle is missing', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.findOne.mockResolvedValue(entity('db'));
+			dbStore.read.mockResolvedValue(null);
+
+			await executionPersistence.findSingleExecution('exec-1', { includeData: true });
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }),
+			);
+		});
+
+		it('emits a read failure and no write when the read-merge bundle is missing', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.findOne.mockResolvedValue(entity('db'));
+			executionRepository.manager.transaction = createMockTx(createMockTransaction());
+			dbStore.read.mockResolvedValue(null);
+
+			await expect(
+				executionPersistence.updateExistingExecution('exec-1', { data: runData }),
+			).rejects.toBeInstanceOf(MissingExecutionDataError);
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }),
+			);
+			expect(eventService.emit).not.toHaveBeenCalledWith('execution-data-write', expect.anything());
+		});
+
+		it('counts dropped bundles from a partial batch read as unreadable', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.find.mockResolvedValue([entity('db', 'exec-1'), entity('db', 'exec-2')]);
+			dbStore.readMany.mockResolvedValue(new Map([['exec-1', readBundle]]));
+
+			await executionPersistence.findMultipleExecutions({}, { includeData: true });
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: true, unreadableBundles: 1 }),
+			);
+		});
+
+		it('emits a failed read with zero unreadable bundles when the store read throws', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.findOne.mockResolvedValue(entity('db'));
+			dbStore.read.mockRejectedValueOnce(new Error('db connection lost'));
+
+			await expect(
+				executionPersistence.findSingleExecution('exec-1', { includeData: true }),
+			).rejects.toThrow('db connection lost');
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 0 }),
+			);
+		});
+
+		it('emits a failed read with one unreadable bundle when a fetched bundle fails to deserialize', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.findOne.mockResolvedValue(entity('db'));
+			dbStore.read.mockResolvedValue({ ...readBundle, data: 'not-valid-flatted' });
+
+			await expect(
+				executionPersistence.findSingleExecution('exec-1', {
+					includeData: true,
+					unflattenData: true,
+				}),
+			).rejects.toBeInstanceOf(CorruptedExecutionDataError);
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }),
+			);
+		});
+
+		it('emits a failed read with one unreadable bundle when the store read reports corruption', async () => {
+			const executionPersistence = createPersistenceService('fs');
+			executionRepository.findOne.mockResolvedValue(entity('fs'));
+			fsStore.read.mockRejectedValueOnce(
+				new CorruptedExecutionDataError(
+					{ workflowId: 'wf-1', executionId: 'exec-1' },
+					new Error('x'),
+				),
+			);
+
+			await expect(
+				executionPersistence.findSingleExecution('exec-1', { includeData: true }),
+			).rejects.toBeInstanceOf(CorruptedExecutionDataError);
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'fs', success: false, unreadableBundles: 1 }),
+			);
+		});
+
+		it('emits a read failure with one unreadable bundle when the read-merge read reports corruption', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.findOne.mockResolvedValue(entity('db'));
+			executionRepository.manager.transaction = createMockTx(createMockTransaction());
+			dbStore.read.mockRejectedValueOnce(
+				new CorruptedExecutionDataError(
+					{ workflowId: 'wf-1', executionId: 'exec-1' },
+					new Error('x'),
+				),
+			);
+
+			await expect(
+				executionPersistence.updateExistingExecution('exec-1', { data: runData }),
+			).rejects.toBeInstanceOf(CorruptedExecutionDataError);
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }),
+			);
+		});
+
+		it('counts a corrupt bundle in a batch as unreadable, reports it, and returns the rest', async () => {
+			const executionPersistence = createPersistenceService('db');
+			const good = entity('db', 'exec-1');
+			const corrupt = entity('db', 'exec-2');
+			executionRepository.find.mockResolvedValue([good, corrupt]);
+			dbStore.readMany.mockResolvedValue(
+				new Map([
+					['exec-1', readBundle],
+					['exec-2', { ...readBundle, data: 'not-valid-flatted' }],
+				]),
+			);
+
+			const result = await executionPersistence.findMultipleExecutions(
+				{},
+				{ includeData: true, unflattenData: true },
+			);
+
+			expect(result.map((e) => e.id)).toEqual(['exec-1']);
+			expect(executionRepository.reportInvalidExecutions).toHaveBeenCalledWith([corrupt]);
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: true, unreadableBundles: 1 }),
+			);
+		});
+
+		it('fails the batch when an assembly error is not a corrupt bundle', async () => {
+			const executionPersistence = createPersistenceService('db');
+			// `metadata: undefined` makes assembly throw a plain `TypeError`, not a corruption error.
+			const broken = {
+				id: 'exec-1',
+				workflowId: 'wf-1',
+				storedAt: 'db',
+				metadata: undefined,
+				annotation: undefined,
+				status: 'success',
+			} as unknown as ExecutionEntity;
+			executionRepository.find.mockResolvedValue([broken]);
+			dbStore.readMany.mockResolvedValue(new Map([['exec-1', readBundle]]));
+
+			await expect(
+				executionPersistence.findMultipleExecutions({}, { includeData: true }),
+			).rejects.toThrow();
+
+			expect(executionRepository.reportInvalidExecutions).not.toHaveBeenCalled();
+		});
+
+		it('treats an unsupported execution-data version as a corrupt bundle', async () => {
+			const executionPersistence = createPersistenceService('db');
+			executionRepository.findOne.mockResolvedValue(entity('db'));
+			// Parses fine, but carries a version the migration step does not support.
+			dbStore.read.mockResolvedValue({ ...readBundle, data: '[{"version":99}]' });
+
+			await expect(
+				executionPersistence.findSingleExecution('exec-1', {
+					includeData: true,
+					unflattenData: true,
+				}),
+			).rejects.toBeInstanceOf(CorruptedExecutionDataError);
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }),
+			);
+		});
+
+		it('keeps the missing-bundle count when an unexpected error fails the batch', async () => {
+			const executionPersistence = createPersistenceService('db');
+			const missing = entity('db', 'exec-1');
+			// `metadata: undefined` makes assembly throw a plain `TypeError`, not a corruption error.
+			const broken = {
+				id: 'exec-2',
+				workflowId: 'wf-1',
+				storedAt: 'db',
+				metadata: undefined,
+				annotation: undefined,
+				status: 'success',
+			} as unknown as ExecutionEntity;
+			executionRepository.find.mockResolvedValue([missing, broken]);
+			dbStore.readMany.mockResolvedValue(new Map([['exec-2', readBundle]]));
+
+			await expect(
+				executionPersistence.findMultipleExecutions({}, { includeData: true }),
+			).rejects.toThrow();
+
+			expect(eventService.emit).toHaveBeenCalledWith(
+				'execution-data-read',
+				expect.objectContaining({ mode: 'db', success: false, unreadableBundles: 1 }),
+			);
+		});
+	});
 });

packages/cli/src/executions/execution-persistence.ts

@@ -19,6 +19,7 @@ import { BinaryDataService, ErrorReporter, StorageConfig } from 'n8n-core';
 import type { IRunExecutionData, IRunExecutionDataAll } from 'n8n-workflow';
 import { migrateRunExecutionData, UnexpectedError } from 'n8n-workflow';
 
+import { CorruptedExecutionDataError } from './execution-data/corrupted-execution-data.error';
 import { DbStore } from './execution-data/db-store';
 import { FsStore } from './execution-data/fs-store';
 import { MissingExecutionDataError } from './execution-data/missing-execution-data.error';
@@ -29,9 +30,12 @@ import type {
 	WorkflowSnapshot,
 } from './execution-data/types';
 import { DuplicateExecutionError } from '../errors/duplicate-execution.error';
+import { EventService } from '../events/event.service';
 
 type DeletionTarget = ExecutionRef & { storedAt: ExecutionDataStorageLocation };
 
+type FoundExecution = IExecutionFlattedDb | IExecutionResponse | IExecutionBase;
+
 type UpdatableEntityColumns = Omit<
 	Partial<IExecutionResponse>,
 	| 'id'
@@ -59,6 +63,7 @@ export class ExecutionPersistence {
 		private readonly executionsConfig: ExecutionsConfig,
 		private readonly databaseConfig: DatabaseConfig,
 		private readonly errorReporter: ErrorReporter,
+		private readonly eventService: EventService,
 	) {}
 
 	/**
@@ -72,7 +77,6 @@ export class ExecutionPersistence {
 		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;
 
 		try {
@@ -80,9 +84,16 @@ export class ExecutionPersistence {
 				const { identifiers } = await tx.insert(ExecutionEntity, executionEntity);
 				const executionId = String(identifiers[0].id);
 				const ref = { workflowId: id, executionId };
-				const bundle = { data, workflowData: workflowSnapshot, workflowVersionId };
+				const store = this.getStoreFor(storedAt);
 
-				await this.getStoreFor(storedAt).write(ref, bundle, tx);
+				await this.trackWrite(storedAt, async () => {
+					const bundle = {
+						data: stringify(rawData),
+						workflowData: workflowSnapshot,
+						workflowVersionId,
+					};
+					await store.write(ref, bundle, tx);
+				});
 
 				return executionId;
 			});
@@ -120,7 +131,7 @@ export class ExecutionPersistence {
 		const ref = { workflowId: entity.workflowId, executionId };
 		const store = this.getStoreFor(entity.storedAt);
 
-		return await this.applyDataUpdate(ref, store, execution, conditions);
+		return await this.applyDataUpdate(ref, store, entity.storedAt, execution, conditions);
 	}
 
 	/**
@@ -169,7 +180,7 @@ export class ExecutionPersistence {
 			unflattenData?: boolean;
 			where?: FindOptionsWhere<ExecutionEntity>;
 		},
-	): Promise<IExecutionFlattedDb | IExecutionResponse | IExecutionBase | undefined> {
+	): Promise<FoundExecution | undefined> {
 		if (!options?.includeData) {
 			return await this.executionRepository.findSingleExecution(id, options);
 		}
@@ -185,23 +196,35 @@ export class ExecutionPersistence {
 		if (!entity) return undefined;
 
 		const store = this.getStoreFor(entity.storedAt);
-		const bundle = await store.read({ workflowId: entity.workflowId, executionId: entity.id });
+		const ref = { workflowId: entity.workflowId, executionId: entity.id };
 
-		if (!bundle) {
-			if (entity.storedAt === 'db') {
-				this.executionRepository.reportInvalidExecutions([entity]);
-				return undefined;
+		const start = Date.now();
+		let success = false;
+		let unreadableBundles = 0;
+		try {
+			const bundle = await store.read(ref);
+			if (!bundle) {
+				unreadableBundles = 1;
+				if (entity.storedAt === 'db') {
+					this.executionRepository.reportInvalidExecutions([entity]);
+					return undefined;
+				}
+				throw new MissingExecutionDataError(ref);
 			}
-			throw new MissingExecutionDataError({
-				workflowId: entity.workflowId,
-				executionId: entity.id,
+			const assembled = await this.assembleExecution(entity, bundle, options);
+			success = true;
+			return assembled as FoundExecution;
+		} catch (error) {
+			if (error instanceof CorruptedExecutionDataError) unreadableBundles = 1;
+			throw error;
+		} finally {
+			this.eventService.emit('execution-data-read', {
+				mode: entity.storedAt,
+				durationMs: Date.now() - start,
+				success,
+				unreadableBundles,
 			});
 		}
-
-		return (await this.assembleExecution(entity, bundle, options)) as
-			| IExecutionFlattedDb
-			| IExecutionResponse
-			| IExecutionBase;
 	}
 
 	/**
@@ -275,30 +298,60 @@ export class ExecutionPersistence {
 			entitiesByLocation.set(entity.storedAt, group);
 		}
 
-		const bundlesById = new Map<string, ExecutionDataBundle>();
+		const assembledById = new Map<string, Awaited<ReturnType<typeof this.assembleExecution>>>();
 		await Promise.all(
 			[...entitiesByLocation].map(async ([location, group]) => {
 				const refs = group.map((e) => ({ workflowId: e.workflowId, executionId: e.id }));
-				const bundles = await this.getStoreFor(location).readMany(refs);
-				for (const [id, bundle] of bundles) bundlesById.set(id, bundle);
+				const store = this.getStoreFor(location);
+				const start = Date.now();
+				let success = false;
+				let unreadableBundles = 0;
+				try {
+					const bundles = await store.readMany(refs);
+					const missing = group.filter((e) => !bundles.has(e.id));
+					if (missing.length > 0) this.executionRepository.reportInvalidExecutions(missing);
+					unreadableBundles = missing.length;
+
+					const settled = await Promise.allSettled(
+						group.map(async (entity) => {
+							const bundle = bundles.get(entity.id);
+							if (!bundle) return;
+							assembledById.set(entity.id, await this.assembleExecution(entity, bundle, options));
+						}),
+					);
+					const corrupt = group.filter((_, i) => {
+						const outcome = settled[i];
+						return (
+							outcome.status === 'rejected' && outcome.reason instanceof CorruptedExecutionDataError
+						);
+					});
+					unreadableBundles += corrupt.length;
+					if (corrupt.length > 0) this.executionRepository.reportInvalidExecutions(corrupt);
+
+					for (const outcome of settled) {
+						if (
+							outcome.status === 'rejected' &&
+							!(outcome.reason instanceof CorruptedExecutionDataError)
+						) {
+							throw outcome.reason;
+						}
+					}
+
+					success = true;
+				} finally {
+					this.eventService.emit('execution-data-read', {
+						mode: location,
+						durationMs: Date.now() - start,
+						success,
+						unreadableBundles,
+					});
+				}
 			}),
 		);
 
-		// Report invalid entities when they are found to be missing from the stores.
-		const invalidEntities = entities.filter((e) => !bundlesById.has(e.id));
-		if (invalidEntities.length > 0) {
-			this.executionRepository.reportInvalidExecutions(invalidEntities);
-		}
-
-		const assembled = await Promise.all(
-			entities.map(async (entity) => {
-				const bundle = bundlesById.get(entity.id);
-				if (!bundle) return null;
-				return await this.assembleExecution(entity, bundle, options);
-			}),
-		);
-
-		return assembled.filter((e): e is NonNullable<typeof e> => e !== null) as
+		return entities
+			.map((e) => assembledById.get(e.id))
+			.filter((e): e is NonNullable<typeof e> => e !== undefined) as
 			| IExecutionFlattedDb[]
 			| IExecutionResponse[]
 			| IExecutionBase[];
@@ -358,6 +411,7 @@ export class ExecutionPersistence {
 	private async applyDataUpdate(
 		ref: ExecutionRef,
 		store: ExecutionDataStore,
+		mode: ExecutionDataStorageLocation,
 		execution: Partial<IExecutionResponse>,
 		conditions?: UpdateExecutionConditions,
 	): Promise<boolean> {
@@ -385,28 +439,34 @@ export class ExecutionPersistence {
 			}
 
 			if (data !== undefined && workflowData !== undefined && store === this.dbStore) {
-				const result = await tx.update(
-					ExecutionData,
-					{ executionId: ref.executionId },
-					{ data: stringify(data), workflowData: this.toWorkflowSnapshot(workflowData) },
-				);
-				if ((result.affected ?? 0) === 0) throw new MissingExecutionDataError(ref);
+				await this.trackWrite(mode, async () => {
+					const result = await tx.update(
+						ExecutionData,
+						{ executionId: ref.executionId },
+						{ data: stringify(data), workflowData: this.toWorkflowSnapshot(workflowData) },
+					);
+					if ((result.affected ?? 0) === 0) throw new MissingExecutionDataError(ref);
+				});
 				return true;
 			}
 
-			const existing = await store.read(ref, tx);
+			const existing = await this.trackRead(mode, async () => await store.read(ref, tx));
 			if (!existing) throw new MissingExecutionDataError(ref);
 
-			await store.write(
-				ref,
-				{
-					data: data !== undefined ? stringify(data) : existing.data,
-					workflowData: workflowData
-						? this.toWorkflowSnapshot(workflowData)
-						: existing.workflowData,
-					workflowVersionId: existing.workflowVersionId,
-				},
-				tx,
+			await this.trackWrite(
+				mode,
+				async () =>
+					await store.write(
+						ref,
+						{
+							data: data !== undefined ? stringify(data) : existing.data,
+							workflowData: workflowData
+								? this.toWorkflowSnapshot(workflowData)
+								: existing.workflowData,
+							workflowVersionId: existing.workflowVersionId,
+						},
+						tx,
+					),
 			);
 
 			return true;
@@ -461,6 +521,46 @@ export class ExecutionPersistence {
 		return where;
 	}
 
+	private async trackRead<T>(mode: ExecutionDataStorageLocation, op: () => Promise<T>): Promise<T> {
+		const start = Date.now();
+		let success = false;
+		let unreadableBundles = 0;
+		try {
+			const result = await op();
+			success = result !== null && result !== undefined;
+			if (!success) unreadableBundles = 1;
+			return result;
+		} catch (error) {
+			if (error instanceof CorruptedExecutionDataError) unreadableBundles = 1;
+			throw error;
+		} finally {
+			this.eventService.emit('execution-data-read', {
+				mode,
+				durationMs: Date.now() - start,
+				success,
+				unreadableBundles,
+			});
+		}
+	}
+
+	private async trackWrite(
+		mode: ExecutionDataStorageLocation,
+		op: () => Promise<void>,
+	): Promise<void> {
+		const start = Date.now();
+		let success = false;
+		try {
+			await op();
+			success = true;
+		} finally {
+			this.eventService.emit('execution-data-write', {
+				mode,
+				durationMs: Date.now() - start,
+				success,
+			});
+		}
+	}
+
 	private getStoreFor(location: ExecutionDataStorageLocation): ExecutionDataStore {
 		switch (location) {
 			case 'db':
@@ -485,7 +585,8 @@ export class ExecutionPersistence {
 		options: { unflattenData?: boolean; includeAnnotation?: boolean },
 	) {
 		const { metadata, annotation, ...rest } = entity;
-		const data = await this.parseExecutionData(bundle.data, options);
+		const ref = { workflowId: entity.workflowId, executionId: entity.id };
+		const data = await this.parseExecutionData(ref, bundle.data, options);
 		const serializedAnnotation = this.serializeAnnotation(annotation);
 
 		if (entity.status === 'success' && bundle.data === '[]') {
@@ -507,14 +608,19 @@ export class ExecutionPersistence {
 	}
 
 	private async parseExecutionData(
+		ref: ExecutionRef,
 		data: string,
 		options: { unflattenData?: boolean },
 	): Promise<IRunExecutionData | string | undefined> {
 		if (!options.unflattenData) return data;
 
-		const deserialized: unknown = await parseFlatted(data);
-		if (!deserialized) return undefined;
-		return migrateRunExecutionData(deserialized as IRunExecutionDataAll);
+		try {
+			const deserialized: unknown = await parseFlatted(data);
+			if (!deserialized) return undefined;
+			return migrateRunExecutionData(deserialized as IRunExecutionDataAll);
+		} catch (error) {
+			throw new CorruptedExecutionDataError(ref, error);
+		}
 	}
 
 	private serializeAnnotation(annotation: ExecutionEntity['annotation']) {

packages/cli/src/metrics/__tests__/prometheus-metrics.service.test.ts

@@ -6,7 +6,7 @@ import type { WorkflowRepository, LicenseMetricsRepository } from '@n8n/db';
 import type express from 'express';
 import promBundle from 'express-prom-bundle';
 import { mock } from 'jest-mock-extended';
-import type { InstanceSettings } from 'n8n-core';
+import type { InstanceSettings, StorageConfig } from 'n8n-core';
 import { EventMessageTypeNames } from 'n8n-workflow';
 import promClient from 'prom-client';
 
@@ -35,6 +35,7 @@ describe('PrometheusMetricsService', () => {
 	let instanceSettings: InstanceSettings;
 	let workflowRepository: WorkflowRepository;
 	let licenseMetricsRepository: LicenseMetricsRepository;
+	let storageConfig: StorageConfig;
 	let prometheusMetricsService: PrometheusMetricsService;
 
 	beforeEach(() => {
@@ -56,6 +57,7 @@ describe('PrometheusMetricsService', () => {
 					includeWorkflowExecutionDuration: false,
 					includeWorkflowNameLabel: false,
 					includeWorkflowStatistics: false,
+					includeExecutionDataMetrics: false,
 					activeWorkflowCountInterval: 30,
 					workflowStatisticsInterval: 30,
 				},
@@ -78,6 +80,7 @@ describe('PrometheusMetricsService', () => {
 		instanceSettings = mock<InstanceSettings>({ instanceType: 'main' });
 		workflowRepository = mock<WorkflowRepository>();
 		licenseMetricsRepository = mock<LicenseMetricsRepository>();
+		storageConfig = mock<StorageConfig>({ modeTag: 'db' });
 
 		prometheusMetricsService = new PrometheusMetricsService(
 			mock(),
@@ -87,6 +90,7 @@ describe('PrometheusMetricsService', () => {
 			instanceSettings,
 			workflowRepository,
 			licenseMetricsRepository,
+			storageConfig,
 		);
 
 		promClient.Counter.prototype.inc = jest.fn();
@@ -103,6 +107,12 @@ describe('PrometheusMetricsService', () => {
 		prometheusMetricsService.disableAllLabels();
 	});
 
+	/** Capture the handler registered via `eventService.on(eventName, handler)`. */
+	const getEventServiceHandler = (eventName: string) => {
+		const call = (eventService.on as jest.Mock).mock.calls.find((c) => c[0] === eventName);
+		return call ? call[1] : undefined;
+	};
+
 	describe('constructor', () => {
 		it('should enable metrics based on global config', async () => {
 			const customGlobalConfig = { ...globalConfig };
@@ -115,6 +125,7 @@ describe('PrometheusMetricsService', () => {
 				instanceSettings,
 				mock(),
 				mock(),
+				mock(),
 			);
 
 			await customPrometheusMetricsService.init(app);
@@ -826,12 +837,6 @@ describe('PrometheusMetricsService', () => {
 	});
 
 	describe('token exchange metrics', () => {
-		// Helper to capture an eventService.on handler by event name
-		const getEventServiceHandler = (eventName: string) => {
-			const call = (eventService.on as jest.Mock).mock.calls.find((c) => c[0] === eventName);
-			return call ? call[1] : undefined;
-		};
-
 		it('should register all 6 token exchange counters on init', async () => {
 			await prometheusMetricsService.init(app);
 
@@ -972,4 +977,101 @@ describe('PrometheusMetricsService', () => {
 			expect(linkedCounter?.inc).toHaveBeenCalledWith(1);
 		});
 	});
+
+	describe('execution data metrics', () => {
+		it('should register execution data metrics and set a storage-mode gauge when enabled', async () => {
+			prometheusMetricsService.enableMetric('executionData');
+			await prometheusMetricsService.init(app);
+
+			expect(promClient.Counter).toHaveBeenCalledWith({
+				name: 'n8n_execution_data_reads_total',
+				help: 'Total number of execution data reads.',
+				labelNames: ['mode', 'result'],
+			});
+			expect(promClient.Counter).toHaveBeenCalledWith({
+				name: 'n8n_execution_data_writes_total',
+				help: 'Total number of execution data writes.',
+				labelNames: ['mode', 'result'],
+			});
+			expect(promClient.Counter).toHaveBeenCalledWith(
+				expect.objectContaining({ name: 'n8n_execution_data_unreadable_bundles_total' }),
+			);
+			expect(promClient.Histogram).toHaveBeenCalledWith(
+				expect.objectContaining({ name: 'n8n_execution_data_read_duration_seconds' }),
+			);
+			expect(promClient.Histogram).toHaveBeenCalledWith(
+				expect.objectContaining({ name: 'n8n_execution_data_write_duration_seconds' }),
+			);
+			expect(promClient.Gauge).toHaveBeenCalledWith(
+				expect.objectContaining({ name: 'n8n_execution_data_storage_mode', labelNames: ['mode'] }),
+			);
+
+			// @ts-expect-error private field
+			const gauge = prometheusMetricsService.gauges.executionDataStorageMode;
+			expect(gauge.set).toHaveBeenCalledWith({ mode: 'db' }, 0);
+			expect(gauge.set).toHaveBeenCalledWith({ mode: 'fs' }, 0);
+			expect(gauge.set).toHaveBeenCalledWith({ mode: 'db' }, 1); // configured mode (storageConfig.modeTag)
+		});
+
+		it('should not register execution data metrics when disabled', async () => {
+			await prometheusMetricsService.init(app);
+
+			expect(promClient.Counter).not.toHaveBeenCalledWith(
+				expect.objectContaining({ name: 'n8n_execution_data_reads_total' }),
+			);
+		});
+
+		it('should count a read and observe its duration (seconds) on a successful read event', async () => {
+			prometheusMetricsService.enableMetric('executionData');
+			await prometheusMetricsService.init(app);
+
+			getEventServiceHandler('execution-data-read')({
+				mode: 'db',
+				durationMs: 200,
+				success: true,
+				unreadableBundles: 0,
+			});
+
+			// @ts-expect-error private field
+			const reads = prometheusMetricsService.counters.executionDataReadsTotal;
+			expect(reads?.inc).toHaveBeenCalledWith({ mode: 'db', result: 'success' }, 1);
+			// @ts-expect-error private field
+			const histogram = prometheusMetricsService.histograms.executionDataReadDuration;
+			expect(histogram.observe).toHaveBeenCalledWith({ mode: 'db' }, 0.2);
+		});
+
+		it('should count unreadable bundles reported by a read event', async () => {
+			prometheusMetricsService.enableMetric('executionData');
+			await prometheusMetricsService.init(app);
+
+			getEventServiceHandler('execution-data-read')({
+				mode: 'fs',
+				durationMs: 5,
+				success: true,
+				unreadableBundles: 3,
+			});
+
+			// @ts-expect-error private field
+			const unreadable = prometheusMetricsService.counters.executionDataUnreadableBundlesTotal;
+			expect(unreadable?.inc).toHaveBeenCalledWith({ mode: 'fs' }, 3);
+		});
+
+		it('should count a failed write without observing its duration', async () => {
+			prometheusMetricsService.enableMetric('executionData');
+			await prometheusMetricsService.init(app);
+
+			getEventServiceHandler('execution-data-write')({
+				mode: 'db',
+				durationMs: 10,
+				success: false,
+			});
+
+			// @ts-expect-error private field
+			const writes = prometheusMetricsService.counters.executionDataWritesTotal;
+			expect(writes?.inc).toHaveBeenCalledWith({ mode: 'db', result: 'failure' }, 1);
+			// @ts-expect-error private field
+			const histogram = prometheusMetricsService.histograms.executionDataWriteDuration;
+			expect(histogram.observe).not.toHaveBeenCalled();
+		});
+	});
 });

packages/cli/src/metrics/__tests__/prometheus-metrics.service.unmocked.test.ts

@@ -3,7 +3,7 @@ import { GlobalConfig } from '@n8n/config';
 import type { WorkflowRepository, LicenseMetricsRepository } from '@n8n/db';
 import type express from 'express';
 import { mock } from 'jest-mock-extended';
-import type { InstanceSettings } from 'n8n-core';
+import type { InstanceSettings, StorageConfig } from 'n8n-core';
 import promClient from 'prom-client';
 
 import { EventMessageWorkflow } from '@/eventbus/event-message-classes/event-message-workflow';
@@ -21,6 +21,7 @@ const cacheService = mock<CacheService>();
 const eventService = mock<EventService>();
 const instanceSettings = mock<InstanceSettings>({ instanceType: 'main' });
 const workflowRepository = mock<WorkflowRepository>();
+const storageConfig = mock<StorageConfig>({ modeTag: 'db' });
 const app = mock<express.Application>();
 const eventBus = new MessageEventBus(mock(), mock(), mock(), mock(), mock(), mock());
 
@@ -46,6 +47,7 @@ describe('workflow_success_total', () => {
 			instanceSettings,
 			workflowRepository,
 			mock<LicenseMetricsRepository>(),
+			storageConfig,
 		);
 
 		await prometheusMetricsService.init(app);
@@ -90,6 +92,7 @@ workflow_success_total{workflow_id="1234"} 1"
 			instanceSettings,
 			workflowRepository,
 			mock<LicenseMetricsRepository>(),
+			storageConfig,
 		);
 
 		await prometheusMetricsService.init(app);
@@ -131,6 +134,7 @@ workflow_success_total{workflow_name="wf_1234"} 1"
 			instanceSettings,
 			workflowRepository,
 			mock<LicenseMetricsRepository>(),
+			storageConfig,
 		);
 
 		// ACT
@@ -176,6 +180,7 @@ describe('Active workflow count', () => {
 		instanceSettings,
 		workflowRepository,
 		mock<LicenseMetricsRepository>(),
+		storageConfig,
 	);
 
 	afterEach(() => {

packages/cli/src/metrics/prometheus-metrics.service.ts

@@ -8,7 +8,7 @@ import { Service } from '@n8n/di';
 import type express from 'express';
 import promBundle from 'express-prom-bundle';
 import { DateTime } from 'luxon';
-import { InstanceSettings } from 'n8n-core';
+import { InstanceSettings, StorageConfig } from 'n8n-core';
 import { EventMessageTypeNames, jsonParse } from 'n8n-workflow';
 import promClient, { type Counter, type Gauge, type Histogram } from 'prom-client';
 import semverParse from 'semver/functions/parse';
@@ -21,6 +21,11 @@ import { CacheService } from '@/services/cache/cache.service';
 
 import type { Includes, MetricCategory, MetricLabel } from './types';
 
+/** Bucket boundaries (in seconds) shared by all duration histograms. */
+const DURATION_BUCKETS_SECONDS = [
+	0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600,
+];
+
 @Service()
 export class PrometheusMetricsService {
 	constructor(
@@ -31,12 +36,15 @@ export class PrometheusMetricsService {
 		private readonly instanceSettings: InstanceSettings,
 		private readonly workflowRepository: WorkflowRepository,
 		private readonly licenseMetricsRepository: LicenseMetricsRepository,
+		private readonly storageConfig: StorageConfig,
 	) {}
 
 	private readonly counters: { [key: string]: Counter<string> | null } = {};
 
 	private tokenExchangeListenersRegistered = false;
 
+	private executionDataListenersRegistered = false;
+
 	private readonly gauges: Record<string, Gauge<string>> = {};
 
 	private readonly histograms: Record<string, Histogram<string>> = {};
@@ -53,6 +61,7 @@ export class PrometheusMetricsService {
 			workflowExecutionDuration:
 				this.globalConfig.endpoints.metrics.includeWorkflowExecutionDuration,
 			workflowStatistics: this.globalConfig.endpoints.metrics.includeWorkflowStatistics,
+			executionData: this.globalConfig.endpoints.metrics.includeExecutionDataMetrics,
 		},
 		labels: {
 			credentialsType: this.globalConfig.endpoints.metrics.includeCredentialTypeLabel,
@@ -79,6 +88,7 @@ export class PrometheusMetricsService {
 		this.initActiveWorkflowCountMetric();
 		this.initWorkflowStatisticsMetrics();
 		this.initTokenExchangeMetrics();
+		this.initExecutionDataMetrics();
 		this.mountMetricsEndpoint(app);
 	}
 
@@ -367,7 +377,7 @@ export class PrometheusMetricsService {
 			name: this.prefix + 'workflow_execution_duration_seconds',
 			help: 'Workflow execution duration in seconds.',
 			labelNames,
-			buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600],
+			buckets: DURATION_BUCKETS_SECONDS,
 		});
 
 		this.eventService.on('workflow-post-execute', ({ runData, workflow }) => {
@@ -712,4 +722,84 @@ export class PrometheusMetricsService {
 			this.counters.tokenExchangeIdentityLinkedTotal?.inc(1);
 		});
 	}
+
+	private initExecutionDataMetrics() {
+		if (!this.includes.metrics.executionData) return;
+
+		const readsTotal = (this.counters.executionDataReadsTotal = new promClient.Counter({
+			name: this.prefix + 'execution_data_reads_total',
+			help: 'Total number of execution data reads.',
+			labelNames: ['mode', 'result'],
+		}));
+
+		const writesTotal = (this.counters.executionDataWritesTotal = new promClient.Counter({
+			name: this.prefix + 'execution_data_writes_total',
+			help: 'Total number of execution data writes.',
+			labelNames: ['mode', 'result'],
+		}));
+
+		const unreadableBundlesTotal = (this.counters.executionDataUnreadableBundlesTotal =
+			new promClient.Counter({
+				name: this.prefix + 'execution_data_unreadable_bundles_total',
+				help: 'Total number of execution data bundles that were missing or corrupt on read.',
+				labelNames: ['mode'],
+			}));
+
+		for (const mode of ['db', 'fs'] as const) {
+			for (const result of ['success', 'failure'] as const) {
+				readsTotal.inc({ mode, result }, 0);
+				writesTotal.inc({ mode, result }, 0);
+			}
+			unreadableBundlesTotal.inc({ mode }, 0);
+		}
+
+		this.histograms.executionDataReadDuration = new promClient.Histogram({
+			name: this.prefix + 'execution_data_read_duration_seconds',
+			help: 'Execution data read duration in seconds (fetch + deserialize).',
+			labelNames: ['mode'],
+			buckets: DURATION_BUCKETS_SECONDS,
+		});
+
+		this.histograms.executionDataWriteDuration = new promClient.Histogram({
+			name: this.prefix + 'execution_data_write_duration_seconds',
+			help: 'Execution data write duration in seconds.',
+			labelNames: ['mode'],
+			buckets: DURATION_BUCKETS_SECONDS,
+		});
+
+		this.gauges.executionDataStorageMode = new promClient.Gauge({
+			name: this.prefix + 'execution_data_storage_mode',
+			help: 'Configured execution data storage mode (1 for the active mode, 0 otherwise).',
+			labelNames: ['mode'],
+		});
+		this.gauges.executionDataStorageMode.set({ mode: 'db' }, 0);
+		this.gauges.executionDataStorageMode.set({ mode: 'fs' }, 0);
+		this.gauges.executionDataStorageMode.set({ mode: this.storageConfig.modeTag }, 1);
+
+		if (this.executionDataListenersRegistered) return;
+		this.executionDataListenersRegistered = true;
+
+		this.eventService.on(
+			'execution-data-read',
+			({ mode, durationMs, success, unreadableBundles }) => {
+				this.counters.executionDataReadsTotal?.inc(
+					{ mode, result: success ? 'success' : 'failure' },
+					1,
+				);
+				if (success)
+					this.histograms.executionDataReadDuration?.observe({ mode }, durationMs / 1000);
+				if (unreadableBundles > 0) {
+					this.counters.executionDataUnreadableBundlesTotal?.inc({ mode }, unreadableBundles);
+				}
+			},
+		);
+
+		this.eventService.on('execution-data-write', ({ mode, durationMs, success }) => {
+			this.counters.executionDataWritesTotal?.inc(
+				{ mode, result: success ? 'success' : 'failure' },
+				1,
+			);
+			if (success) this.histograms.executionDataWriteDuration?.observe({ mode }, durationMs / 1000);
+		});
+	}
 }

packages/cli/src/metrics/types.ts

@@ -5,7 +5,8 @@ export type MetricCategory =
 	| 'logs'
 	| 'queue'
 	| 'workflowExecutionDuration'
-	| 'workflowStatistics';
+	| 'workflowStatistics'
+	| 'executionData';
 
 export type MetricLabel =
 	| 'credentialsType'

packages/cli/test/integration/prometheus-metrics.test.ts

@@ -42,6 +42,7 @@ globalConfig.endpoints.metrics = {
 	activeWorkflowCountInterval: 60,
 	includeWorkflowStatistics: false,
 	workflowStatisticsInterval: 300,
+	includeExecutionDataMetrics: false,
 };
 
 const server = setupTestServer({ endpointGroups: ['metrics'] });
@@ -489,4 +490,23 @@ describe('PrometheusMetricsService', () => {
 		expect(lines).not.toContainEqual(expect.stringContaining('n8n_test_workflows'));
 		expect(lines).not.toContainEqual(expect.stringContaining('n8n_test_credentials'));
 	});
+
+	it('should return execution data metrics if enabled', async () => {
+		prometheusService.enableMetric('executionData');
+		await prometheusService.init(server.app);
+
+		const response = await agent.get('/metrics');
+
+		expect(response.status).toEqual(200);
+
+		const lines = toLines(response);
+
+		expect(lines).toContain('n8n_test_execution_data_reads_total{mode="db",result="success"} 0');
+		expect(lines).toContain('n8n_test_execution_data_writes_total{mode="fs",result="failure"} 0');
+		expect(lines).toContain('n8n_test_execution_data_unreadable_bundles_total{mode="db"} 0');
+
+		expect(
+			lines.some((l) => /^n8n_test_execution_data_storage_mode\{mode="(db|fs)"\} 1$/.test(l)),
+		).toBe(true);
+	});
 });