refactor(core): Decouple DB from blob store interface (#33563)

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

@@ -240,7 +240,6 @@ describe('ExecutionPersistence', () => {
 						workflowData: expectedWorkflowSnapshot,
 						workflowVersionId: 'version-abc',
 					}),
-					mockTx,
 				);
 			});
 
@@ -721,7 +720,6 @@ describe('ExecutionPersistence', () => {
 						// from the entity row, not the incoming workflowData.versionId
 						workflowVersionId: 'v-entity',
 					}),
-					mockTx,
 				);
 				expect(fsStore.read).not.toHaveBeenCalled();
 			});
@@ -751,7 +749,6 @@ describe('ExecutionPersistence', () => {
 				expect(fsStore.write).toHaveBeenCalledWith(
 					{ workflowId, executionId },
 					expect.objectContaining({ workflowVersionId: existingBundle.workflowVersionId }),
-					mockTx,
 				);
 			});
 
@@ -772,7 +769,6 @@ describe('ExecutionPersistence', () => {
 						workflowData: existingBundle.workflowData,
 						workflowVersionId: existingBundle.workflowVersionId,
 					}),
-					mockTx,
 				);
 			});
 
@@ -2220,7 +2216,6 @@ describe('ExecutionPersistence', () => {
 			expect(s3Store.write).toHaveBeenCalledWith(
 				{ workflowId: 'workflow-123', executionId: 'exec-1' },
 				expect.objectContaining({ workflowVersionId: 'version-abc' }),
-				mockTx,
 			);
 			expect(dbStore.write).not.toHaveBeenCalled();
 			expect(fsStore.write).not.toHaveBeenCalled();
@@ -2330,7 +2325,6 @@ describe('ExecutionPersistence', () => {
 			expect(azStore.write).toHaveBeenCalledWith(
 				{ workflowId: 'workflow-123', executionId: 'exec-1' },
 				expect.objectContaining({ workflowVersionId: 'version-abc' }),
-				mockTx,
 			);
 			expect(dbStore.write).not.toHaveBeenCalled();
 			expect(fsStore.write).not.toHaveBeenCalled();

packages/cli/src/executions/execution-data/__tests__/db-store.integration.test.ts

@@ -1,5 +1,6 @@
 import { createWorkflow, testDb } from '@n8n/backend-test-utils';
 import { ExecutionDataRepository, ExecutionRepository } from '@n8n/db';
+import type { EntityManager } from '@n8n/db';
 import { Container } from '@n8n/di';
 
 import { DbStore } from '../db-store';
@@ -11,6 +12,7 @@ import { payload, workflowId } from './mocks';
 let dbStore: DbStore;
 let repository: ExecutionDataRepository;
 let executionRepository: ExecutionRepository;
+let em: EntityManager;
 
 async function createExecution() {
 	const execution = await executionRepository.save({
@@ -30,6 +32,7 @@ beforeAll(async () => {
 	await testDb.init();
 	repository = Container.get(ExecutionDataRepository);
 	executionRepository = Container.get(ExecutionRepository);
+	em = executionRepository.manager;
 	dbStore = Container.get(DbStore);
 	await createWorkflow({ id: workflowId });
 });
@@ -47,7 +50,7 @@ describe('write', () => {
 		const execution = await createExecution();
 		const ref = createExecutionRef(workflowId, execution.id);
 
-		await dbStore.write(ref, payload);
+		await dbStore.write(ref, payload, em);
 
 		const stored = await repository.findOneBy({ executionId: execution.id });
 
@@ -58,7 +61,7 @@ describe('write', () => {
 		const execution = await createExecution();
 		const ref = createExecutionRef(workflowId, execution.id);
 
-		const bytes = await dbStore.write(ref, payload);
+		const bytes = await dbStore.write(ref, payload, em);
 
 		const expected =
 			Buffer.byteLength(payload.data, 'utf8') +
@@ -71,14 +74,14 @@ describe('write', () => {
 		const execution = await createExecution();
 		const ref = createExecutionRef(workflowId, execution.id);
 
-		await dbStore.write(ref, payload);
+		await dbStore.write(ref, payload, em);
 
 		const updatedPayload: ExecutionDataPayload = {
 			...payload,
 			data: '[[{"json":{"updated":true}},null]]',
 		};
 
-		await dbStore.write(ref, updatedPayload);
+		await dbStore.write(ref, updatedPayload, em);
 
 		const count = await repository.count();
 		expect(count).toBe(1);
@@ -92,13 +95,13 @@ describe('overwrite', () => {
 	it('should replace data and snapshot in place and return the byte size', async () => {
 		const execution = await createExecution();
 		const ref = createExecutionRef(workflowId, execution.id);
-		await dbStore.write(ref, payload);
+		await dbStore.write(ref, payload, em);
 
 		const updatedPayload: ExecutionDataPayload = {
 			...payload,
 			data: '[[{"json":{"updated":true}},null]]',
 		};
-		const bytes = await dbStore.overwrite(ref, updatedPayload);
+		const bytes = await dbStore.overwrite(ref, updatedPayload, em);
 
 		const expected =
 			Buffer.byteLength(updatedPayload.data, 'utf8') +
@@ -113,7 +116,7 @@ describe('overwrite', () => {
 	it('should throw MissingExecutionDataError when the row does not exist', async () => {
 		const ref = createExecutionRef(workflowId, '999');
 
-		await expect(dbStore.overwrite(ref, payload)).rejects.toThrow(MissingExecutionDataError);
+		await expect(dbStore.overwrite(ref, payload, em)).rejects.toThrow(MissingExecutionDataError);
 	});
 });
 
@@ -122,11 +125,11 @@ describe('read', () => {
 		const execution = await createExecution();
 		const ref = createExecutionRef(workflowId, execution.id);
 
-		await dbStore.write(ref, payload);
+		await dbStore.write(ref, payload, em);
 
 		const stored = await dbStore.read(ref);
 
-		expect(stored).toMatchObject({ ...payload, version: 1 });
+		expect(stored).toMatchObject(payload);
 	});
 
 	it('should return `null` for non-existent execution', async () => {
@@ -137,23 +140,23 @@ describe('read', () => {
 });
 
 describe('readMany', () => {
-	it('should return a map of bundles keyed by executionId, omitting missing ones', async () => {
+	it('should return a map of payloads keyed by executionId, omitting missing ones', async () => {
 		const [a, b] = await Promise.all([createExecution(), createExecution()]);
 		const refA = createExecutionRef(workflowId, a.id);
 		const refB = createExecutionRef(workflowId, b.id);
-		await dbStore.write(refA, payload);
+		await dbStore.write(refA, payload, em);
 
-		const bundles = await dbStore.readMany([refA, refB]);
+		const payloads = await dbStore.readMany([refA, refB]);
 
-		expect(bundles.size).toBe(1);
-		expect(bundles.get(a.id)).toMatchObject({ ...payload, version: 1 });
-		expect(bundles.has(b.id)).toBe(false);
+		expect(payloads.size).toBe(1);
+		expect(payloads.get(a.id)).toMatchObject(payload);
+		expect(payloads.has(b.id)).toBe(false);
 	});
 
 	it('should return an empty map for an empty array', async () => {
-		const bundles = await dbStore.readMany([]);
+		const payloads = await dbStore.readMany([]);
 
-		expect(bundles.size).toBe(0);
+		expect(payloads.size).toBe(0);
 	});
 
 	it('should batch the IN-clause so a large id set stays within the DB parameter limit', async () => {
@@ -165,48 +168,24 @@ describe('readMany', () => {
 	});
 });
 
-describe('delete', () => {
-	it('should delete data for single execution', async () => {
+describe('readWorkflowData', () => {
+	it('should retrieve only the workflow snapshot', async () => {
 		const execution = await createExecution();
 		const ref = createExecutionRef(workflowId, execution.id);
 
-		await dbStore.write(ref, payload);
+		await dbStore.write(ref, payload, em);
 
-		await dbStore.delete(ref);
+		const snapshot = await dbStore.readWorkflowData(ref);
+
+		expect(snapshot).toMatchObject({
+			workflowData: payload.workflowData,
+			workflowVersionId: payload.workflowVersionId,
+		});
+	});
+
+	it('should return `null` for non-existent execution', async () => {
+		const result = await dbStore.readWorkflowData(createExecutionRef(workflowId, '999'));
 
-		const result = await repository.findOneBy({ executionId: execution.id });
 		expect(result).toBeNull();
 	});
-
-	it('should delete data for multiple executions', async () => {
-		const executions = await Promise.all([createExecution(), createExecution(), createExecution()]);
-
-		const refs = executions.map((e) => createExecutionRef(workflowId, e.id));
-
-		for (const ref of refs) {
-			await dbStore.write(ref, payload);
-		}
-
-		await dbStore.delete([refs[0], refs[1]]);
-
-		const remaining = await repository.find();
-		expect(remaining).toHaveLength(1);
-		expect(remaining[0].executionId).toBe(executions[2].id);
-	});
-
-	it('should skip deletion on empty array', async () => {
-		const execution = await createExecution();
-		const ref = createExecutionRef(workflowId, execution.id);
-
-		await dbStore.write(ref, payload);
-
-		await dbStore.delete([]);
-
-		const count = await repository.count();
-		expect(count).toBe(1);
-	});
-
-	it('should not throw on deleting a non-existent execution', async () => {
-		await expect(dbStore.delete(createExecutionRef(workflowId, '999'))).resolves.toBeUndefined();
-	});
 });

packages/cli/src/executions/execution-data/db-store.ts

@@ -3,43 +3,38 @@ import type { EntityManager } from '@n8n/db';
 import { Service } from '@n8n/di';
 import chunk from 'lodash/chunk';
 
-import { EXECUTION_DATA_BUNDLE_VERSION } from './constants';
 import { MissingExecutionDataError } from './missing-execution-data.error';
-import type {
-	ExecutionDataStore,
-	ExecutionRef,
-	ExecutionDataPayload,
-	ExecutionDataBundle,
-} from './types';
+import type { ExecutionRef, ExecutionDataPayload, BundleWorkflowSnapshot } from './types';
 
 // Max number of ids per IN-clause. Conservative, as some databases cap near 1000.
 const MAX_READ_BATCH_SIZE = 900;
 
 @Service()
-export class DbStore implements ExecutionDataStore {
+export class DbStore {
 	constructor(private readonly repository: ExecutionDataRepository) {}
 
 	async write(
 		{ executionId }: ExecutionRef,
 		payload: ExecutionDataPayload,
-		tx?: EntityManager,
+		tx: EntityManager,
 	): Promise<number> {
-		const repo = this.getRepository(tx);
+		const repo = tx.getRepository(ExecutionData);
 		await repo.upsert({ ...payload, executionId }, ['executionId']);
 		return this.measureBundleBytes(payload);
 	}
 
 	/**
-	 * Overwrite an existing bundle's data and workflow snapshot in place, returning its byte size.
+	 * Overwrite an existing row's data and workflow snapshot in place, returning its byte size.
 	 * Unlike {@link write} (an upsert), the row must already exist - a missing row means the
-	 * execution's data was lost, which we surface rather than silently recreate.
+	 * execution's data was lost, which we surface rather than silently recreate. The
+	 * `workflowVersionId` column is left untouched.
 	 */
 	async overwrite(
 		ref: ExecutionRef,
 		payload: ExecutionDataPayload,
-		tx?: EntityManager,
+		tx: EntityManager,
 	): Promise<number> {
-		const repo = this.getRepository(tx);
+		const repo = tx.getRepository(ExecutionData);
 		const result = await repo.update(
 			{ executionId: ref.executionId },
 			{ data: payload.data, workflowData: payload.workflowData },
@@ -49,7 +44,7 @@ export class DbStore implements ExecutionDataStore {
 	}
 
 	/**
-	 * Byte size of a bundle as stored in the DB: the run data string, the JSON-serialized workflow
+	 * Byte size of a payload as stored in the DB: the run data string, the JSON-serialized workflow
 	 * snapshot, and the version id.
 	 */
 	private measureBundleBytes(payload: ExecutionDataPayload): number {
@@ -63,21 +58,18 @@ export class DbStore implements ExecutionDataStore {
 	async read(
 		{ executionId }: ExecutionRef,
 		tx?: EntityManager,
-	): Promise<ExecutionDataBundle | null> {
-		const repo = this.getRepository(tx);
+	): Promise<ExecutionDataPayload | null> {
+		const repo = tx ? tx.getRepository(ExecutionData) : this.repository;
 		const result = await repo.findOne({
 			where: { executionId },
 			select: ['data', 'workflowData', 'workflowVersionId'],
 		});
 
-		if (!result) return null;
-
-		return { ...result, version: EXECUTION_DATA_BUNDLE_VERSION };
+		return result;
 	}
 
-	async readWorkflowData({ executionId }: ExecutionRef, tx?: EntityManager) {
-		const repo = this.getRepository(tx);
-		const result = await repo.findOne({
+	async readWorkflowData({ executionId }: ExecutionRef): Promise<BundleWorkflowSnapshot | null> {
+		const result = await this.repository.findOne({
 			where: { executionId },
 			select: ['workflowData', 'workflowVersionId'],
 		});
@@ -87,11 +79,10 @@ export class DbStore implements ExecutionDataStore {
 		return { workflowData: result.workflowData, workflowVersionId: result.workflowVersionId };
 	}
 
-	async getDataByteSize({ executionId }: ExecutionRef, tx?: EntityManager): Promise<number | null> {
-		const repo = this.getRepository(tx);
+	async getDataByteSize({ executionId }: ExecutionRef): Promise<number | null> {
 		// LENGTH() avoids loading the column into memory. It counts characters, not bytes, but the
 		// data is a near-ASCII flatted string, so it's close enough to guard on.
-		const row = await repo
+		const row = await this.repository
 			.createQueryBuilder('d')
 			.select('LENGTH(d.data)', 'size')
 			.where('d.executionId = :executionId', { executionId })
@@ -103,8 +94,8 @@ export class DbStore implements ExecutionDataStore {
 	}
 
 	async readMany(refs: ExecutionRef[]) {
-		const bundles = new Map<string, ExecutionDataBundle>();
-		if (refs.length === 0) return bundles;
+		const payloads = new Map<string, ExecutionDataPayload>();
+		if (refs.length === 0) return payloads;
 
 		const ids = refs.map((r) => r.executionId);
 
@@ -117,25 +108,14 @@ export class DbStore implements ExecutionDataStore {
 			});
 
 			for (const row of rows) {
-				bundles.set(row.executionId, {
+				payloads.set(row.executionId, {
 					data: row.data,
 					workflowData: row.workflowData,
 					workflowVersionId: row.workflowVersionId,
-					version: EXECUTION_DATA_BUNDLE_VERSION,
 				});
 			}
 		}
 
-		return bundles;
-	}
-
-	async delete(ref: ExecutionRef | ExecutionRef[]) {
-		const ids = (Array.isArray(ref) ? ref : [ref]).map((r) => r.executionId);
-
-		await this.repository.deleteMany(ids);
-	}
-
-	private getRepository(tx?: EntityManager) {
-		return tx ? tx.getRepository(ExecutionData) : this.repository;
+		return payloads;
 	}
 }

packages/cli/src/executions/execution-data/types.ts

@@ -1,6 +1,9 @@
-import type { EntityManager } from '@n8n/db';
+import type { ExecutionDataStorageLocation } from '@n8n/db';
 import type { IWorkflowBase } from 'n8n-workflow';
 
+/** Storage locations served by {@link ExecutionDataStore} implementations. `db` is handled natively by `DbStore`. */
+export type BlobStorageLocation = Exclude<ExecutionDataStorageLocation, 'db'>;
+
 export type ExecutionRef = {
 	workflowId: string;
 	executionId: string;
@@ -25,34 +28,22 @@ export type ExecutionDataBundle = ExecutionDataPayload & {
 	version: 1;
 };
 
-/** The workflow-snapshot part of a bundle, without the run data. */
+/** The workflow-snapshot part of a payload, without the run data. */
 export type BundleWorkflowSnapshot = Pick<
-	ExecutionDataBundle,
+	ExecutionDataPayload,
 	'workflowData' | 'workflowVersionId'
 >;
 
 /**
- * Persistence operations for execution data bundles. Methods which accept an
- * optional `tx` (`EntityManager`) do so for transactional participation:
- * `DbStore` uses it; `FsStore` ignores it (the filesystem is not transactional).
+ * Persistence operations for execution data bundles kept in
+ * blob storage (filesystem, S3, Azure Blob Storage).
  */
 export interface ExecutionDataStore {
 	init?(): Promise<void>;
 	/** Persist a bundle and return the number of bytes it occupies in this store. */
-	write(ref: ExecutionRef, payload: ExecutionDataPayload, tx?: EntityManager): Promise<number>;
-	read(ref: ExecutionRef, tx?: EntityManager): Promise<ExecutionDataBundle | null>;
+	write(ref: ExecutionRef, payload: ExecutionDataPayload): Promise<number>;
+	read(ref: ExecutionRef): Promise<ExecutionDataBundle | null>;
 	/** Read multiple bundles by ref. Returns a map keyed by `executionId`; missing entries are omitted. */
 	readMany(refs: ExecutionRef[]): Promise<Map<string, ExecutionDataBundle>>;
-	/**
-	 * Read only the workflow snapshot, skipping the run data. Optional — only stores that keep the
-	 * two separately (DB columns) implement it; blob stores omit it and callers fall back.
-	 */
-	readWorkflowData?(ref: ExecutionRef, tx?: EntityManager): Promise<BundleWorkflowSnapshot | null>;
-	/**
-	 * Cheap size of the stored run data, without loading it. Lets the size guard reject legacy
-	 * rows (`jsonSizeBytes === 0`) before reading. Optional — stores that can't size without
-	 * loading omit it, and callers fall back to measuring after read.
-	 */
-	getDataByteSize?(ref: ExecutionRef, tx?: EntityManager): Promise<number | null>;
 	delete(ref: ExecutionRef | ExecutionRef[]): Promise<void>;
 }

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

@@ -3,6 +3,7 @@ import { DatabaseConfig, ExecutionsConfig } from '@n8n/config';
 import { Time } from '@n8n/constants';
 import type {
 	CreateExecutionPayload,
+	EntityManager,
 	ExecutionDataStorageLocation,
 	ExecutionDeletionCriteria,
 	FindManyOptions,
@@ -28,8 +29,8 @@ import { DbStore } from './execution-data/db-store';
 import { FsStore } from './execution-data/fs-store';
 import { MissingExecutionDataError } from './execution-data/missing-execution-data.error';
 import type {
+	BlobStorageLocation,
 	BundleWorkflowSnapshot,
-	ExecutionDataBundle,
 	ExecutionDataPayload,
 	ExecutionDataStore,
 	ExecutionRef,
@@ -111,7 +112,6 @@ export class ExecutionPersistence {
 				const { identifiers } = await tx.insert(ExecutionEntity, executionEntity);
 				const executionId = String(identifiers[0].id);
 				const ref = { workflowId: id, executionId };
-				const store = this.getStoreFor(storedAt);
 
 				const jsonSizeBytes = await this.trackWrite(storedAt, async () => {
 					const bundle: ExecutionDataPayload = {
@@ -119,7 +119,7 @@ export class ExecutionPersistence {
 						workflowData: workflowSnapshot,
 						workflowVersionId,
 					};
-					return await store.write(ref, bundle, tx);
+					return await this.writeData(storedAt, ref, bundle, tx);
 				});
 				const binaryDataSizeBytes = sumBinaryDataBytes(rawData);
 				await tx.update(
@@ -162,11 +162,9 @@ export class ExecutionPersistence {
 		if (!entity) return false;
 
 		const ref = { workflowId: entity.workflowId, executionId };
-		const store = this.getStoreFor(entity.storedAt);
 
 		return await this.applyDataUpdate(
 			ref,
-			store,
 			entity.storedAt,
 			entity.workflowVersionId,
 			execution,
@@ -241,18 +239,18 @@ export class ExecutionPersistence {
 		if (!entity) return undefined;
 
 		const max = this.maxDisplayDataSize(options);
-		const store = this.getStoreFor(entity.storedAt);
 		const ref = { workflowId: entity.workflowId, executionId: entity.id };
 
 		// Over the limit: skip reading run data, loading only the workflow snapshot. Size is known
-		// from `jsonSizeBytes`, or (legacy rows where it's 0) queried cheaply from the store.
+		// from `jsonSizeBytes`, or (legacy db rows where it's 0) queried cheaply from the DB. Blob
+		// stores can't size without loading, so their legacy rows are measured after read instead.
 		if (this.isKnownOversize(entity, max)) {
-			return (await this.assembleSkippedExecution(entity, store, ref, options)) as FoundExecution;
+			return (await this.assembleSkippedExecution(entity, ref, options)) as FoundExecution;
 		}
-		if (max > 0 && entity.jsonSizeBytes === 0) {
-			const size = await store.getDataByteSize?.(ref);
-			if (typeof size === 'number' && size > max) {
-				return (await this.assembleSkippedExecution(entity, store, ref, options)) as FoundExecution;
+		if (max > 0 && entity.jsonSizeBytes === 0 && entity.storedAt === 'db') {
+			const size = await this.dbStore.getDataByteSize(ref);
+			if (size !== null && size > max) {
+				return (await this.assembleSkippedExecution(entity, ref, options)) as FoundExecution;
 			}
 		}
 
@@ -260,7 +258,7 @@ export class ExecutionPersistence {
 		let success = false;
 		let unreadableBundles = 0;
 		try {
-			const bundle = await store.read(ref);
+			const bundle = await this.readData(entity.storedAt, ref);
 			if (!bundle) {
 				unreadableBundles = 1;
 				if (entity.storedAt === 'db') {
@@ -370,12 +368,14 @@ export class ExecutionPersistence {
 		await Promise.all(
 			[...entitiesByLocation].map(async ([location, group]) => {
 				const refs = group.map((e) => ({ workflowId: e.workflowId, executionId: e.id }));
-				const store = this.getStoreFor(location);
 				const start = Date.now();
 				let success = false;
 				let unreadableBundles = 0;
 				try {
-					const bundles = await store.readMany(refs);
+					const bundles =
+						location === 'db'
+							? await this.dbStore.readMany(refs)
+							: await this.getStoreFor(location).readMany(refs);
 					const missing = group.filter((e) => !bundles.has(e.id));
 					if (missing.length > 0) this.executionRepository.reportInvalidExecutions(missing);
 					unreadableBundles = missing.length;
@@ -593,7 +593,6 @@ export class ExecutionPersistence {
 
 	private async applyDataUpdate(
 		ref: ExecutionRef,
-		store: ExecutionDataStore,
 		mode: ExecutionDataStorageLocation,
 		workflowVersionId: string | null,
 		execution: Partial<IExecutionResponse>,
@@ -622,12 +621,12 @@ export class ExecutionPersistence {
 				if (!matchingRow) return false;
 			}
 
-			// Skip the read on a full overwrite. Safe only with a known version id, except for the DB
-			// store: its overwrite leaves that column untouched, whereas others would clobber it with null.
+			// Skip the read on a full overwrite. Safe only with a known version id, except in db mode:
+			// the DB overwrite leaves that column untouched, whereas a blob write would clobber it with null.
 			if (
 				data !== undefined &&
 				workflowData !== undefined &&
-				(workflowVersionId !== null || store === this.dbStore)
+				(workflowVersionId !== null || mode === 'db')
 			) {
 				const binaryDataSizeBytes = sumBinaryDataBytes(data);
 				const jsonSizeBytes = await this.trackWrite(mode, async () => {
@@ -637,9 +636,9 @@ export class ExecutionPersistence {
 						workflowVersionId,
 					};
 
-					return store === this.dbStore
+					return mode === 'db'
 						? await this.dbStore.overwrite(ref, bundle, tx)
-						: await store.write(ref, bundle, tx);
+						: await this.getStoreFor(mode).write(ref, bundle);
 				});
 
 				await tx.update(
@@ -652,7 +651,7 @@ export class ExecutionPersistence {
 
 			// Read the existing bundle to merge the field the caller didn't supply (or to recover the
 			// version id when the entity row doesn't have it).
-			const existing = await this.trackRead(mode, async () => await store.read(ref, tx));
+			const existing = await this.trackRead(mode, async () => await this.readData(mode, ref, tx));
 			if (!existing) throw new MissingExecutionDataError(ref);
 
 			const jsonSizeBytes = await this.trackWrite(mode, async () => {
@@ -664,7 +663,7 @@ export class ExecutionPersistence {
 					workflowVersionId: existing.workflowVersionId,
 				};
 
-				return await store.write(ref, bundle, tx);
+				return await this.writeData(mode, ref, bundle, tx);
 			});
 			// Binary size is derived from the in-memory run data, so only recompute it when the
 			// caller supplied `data`. A workflowData-only update leaves the column untouched (and
@@ -685,9 +684,8 @@ export class ExecutionPersistence {
 	 *
 	 * Stripped fields fall into three categories:
 	 * - **Identity / routing**: `id`, `workflowId` — never updated here.
-	 * - **Stored elsewhere**: `data`, `workflowData` — persisted via the
-	 *   configured {@link ExecutionDataStore} (DB or filesystem), not as columns
-	 *   on the entity row.
+	 * - **Stored elsewhere**: `data`, `workflowData` — persisted per the execution's
+	 *   storage location (DB rows or a blob store), not as columns on the entity row.
 	 * - **Immutable after creation**: `workflowVersionId`, `createdAt`,
 	 *   `startedAt` — set once at insert time and never overwritten.
 	 * - **Not persisted on the entity**: `customData` — handled separately.
@@ -778,10 +776,31 @@ export class ExecutionPersistence {
 		}
 	}
 
-	private getStoreFor(location: ExecutionDataStorageLocation): ExecutionDataStore {
+	/** Write execution data to `mode` storage. In `db` mode, the write participates in `tx`. */
+	private async writeData(
+		mode: ExecutionDataStorageLocation,
+		ref: ExecutionRef,
+		payload: ExecutionDataPayload,
+		tx: EntityManager,
+	): Promise<number> {
+		return mode === 'db'
+			? await this.dbStore.write(ref, payload, tx)
+			: await this.getStoreFor(mode).write(ref, payload);
+	}
+
+	/** Read execution data from `mode` storage. In `db` mode, the read participates in `tx` when given. */
+	private async readData(
+		mode: ExecutionDataStorageLocation,
+		ref: ExecutionRef,
+		tx?: EntityManager,
+	): Promise<ExecutionDataPayload | null> {
+		if (mode !== 'db') return await this.getStoreFor(mode).read(ref);
+
+		return tx ? await this.dbStore.read(ref, tx) : await this.dbStore.read(ref);
+	}
+
+	private getStoreFor(location: BlobStorageLocation): ExecutionDataStore {
 		switch (location) {
-			case 'db':
-				return this.dbStore;
 			case 'fs':
 				return this.fsStore;
 			case 's3':
@@ -812,7 +831,7 @@ export class ExecutionPersistence {
 
 	private async assembleExecution(
 		entity: ExecutionEntity,
-		bundle: ExecutionDataBundle,
+		bundle: ExecutionDataPayload,
 		options: { unflattenData?: boolean; includeAnnotation?: boolean },
 	) {
 		const { metadata, annotation, ...rest } = entity;
@@ -853,14 +872,20 @@ export class ExecutionPersistence {
 		return max > 0 && entity.jsonSizeBytes > 0 && entity.jsonSizeBytes > max;
 	}
 
-	/** Assemble an oversized execution, loading only the workflow snapshot (never the run data). */
+	/**
+	 * Assemble an oversized execution, loading only the workflow snapshot (never the run data).
+	 * Only the DB keeps the snapshot separately from the run data; for blob-stored executions
+	 * we fall back to a stub built from the entity.
+	 */
 	private async assembleSkippedExecution(
 		entity: ExecutionEntity,
-		store: ExecutionDataStore,
 		ref: ExecutionRef,
 		options: { includeAnnotation?: boolean },
 	) {
-		const snapshot = (await store.readWorkflowData?.(ref)) ?? undefined;
+		const snapshot =
+			entity.storedAt === 'db'
+				? ((await this.dbStore.readWorkflowData(ref)) ?? undefined)
+				: undefined;
 		return this.assembleOversizedExecution(
 			entity,
 			{ includeAnnotation: options.includeAnnotation },
@@ -916,10 +941,8 @@ export class ExecutionPersistence {
 		}
 		await Promise.all(
 			oversized.map(async (entity) => {
-				const store = this.getStoreFor(entity.storedAt);
 				const ref = { workflowId: entity.workflowId, executionId: entity.id };
-				const snapshot = (await store.readWorkflowData?.(ref)) ?? undefined;
-				assembledById.set(entity.id, this.assembleOversizedExecution(entity, {}, snapshot));
+				assembledById.set(entity.id, await this.assembleSkippedExecution(entity, ref, {}));
 			}),
 		);
 		return entitiesToRead;
@@ -931,7 +954,7 @@ export class ExecutionPersistence {
 	 */
 	private async assembleReadExecution(
 		entity: ExecutionEntity,
-		bundle: ExecutionDataBundle,
+		bundle: ExecutionDataPayload,
 		options: { unflattenData?: boolean; includeAnnotation?: boolean },
 		max: number,
 	) {