refactor(core): Extend ByteStore for binary payloads (#34654)

packages/@n8n/blob-storage/src/__tests__/azure-byte-store.test.ts

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/unbound-method */
 
-import type { Readable } from 'node:stream';
+import { Readable } from 'node:stream';
 import { mock } from 'vitest-mock-extended';
 
 import type { AzureBlobService } from '../azure-blob/azure-blob.service.ee';
@@ -17,7 +17,7 @@ beforeEach(() => {
 
 describe('write', () => {
 	it('should put the bytes at the key and return the byte count', async () => {
-		const bytes = await store.write('a/b/blob.json', body, 'application/json');
+		const bytes = await store.write('a/b/blob.json', body, { mimeType: 'application/json' });
 		expect(azureBlob.put).toHaveBeenCalledWith('a/b/blob.json', body, {
 			mimeType: 'application/json',
 		});
@@ -30,6 +30,95 @@ describe('write', () => {
 			mimeType: 'application/octet-stream',
 		});
 	});
+
+	it('should forward native metadata', async () => {
+		await store.write('a/b/blob.bin', body, { fileName: 'file.txt', mimeType: 'text/plain' });
+		expect(azureBlob.put).toHaveBeenCalledWith('a/b/blob.bin', body, {
+			fileName: 'file.txt',
+			mimeType: 'text/plain',
+		});
+	});
+
+	it('should buffer a stream before uploading and return the byte count', async () => {
+		const bytes = await store.write('a/b/blob.bin', Readable.from(body));
+		expect(azureBlob.put).toHaveBeenCalledWith('a/b/blob.bin', body, {
+			mimeType: 'application/octet-stream',
+		});
+		expect(bytes).toBe(body.length);
+	});
+});
+
+describe('readStream', () => {
+	it('should return the stored stream, forwarding chunkSize', async () => {
+		const stream = Readable.from(body);
+		azureBlob.get.mockResolvedValueOnce(stream);
+
+		expect(await store.readStream('a/b/blob.json', { chunkSize: 5 })).toBe(stream);
+		expect(azureBlob.get).toHaveBeenCalledWith('a/b/blob.json', { mode: 'stream', chunkSize: 5 });
+	});
+
+	it('should return `null` when the blob is missing', async () => {
+		azureBlob.get.mockRejectedValueOnce(azureNotFoundError);
+		expect(await store.readStream('a/b/blob.json')).toBeNull();
+	});
+
+	it('should rethrow a systemic error', async () => {
+		azureBlob.get.mockRejectedValueOnce(azureThrottledError);
+		await expect(store.readStream('a/b/blob.json')).rejects.toBe(azureThrottledError);
+	});
+
+	it('should reject a fractional chunkSize before downloading', async () => {
+		await expect(store.readStream('a/b/blob.json', { chunkSize: 1.5 })).rejects.toThrow(
+			'positive integer',
+		);
+		expect(azureBlob.get).not.toHaveBeenCalled();
+	});
+});
+
+describe('getMetadata', () => {
+	it('should return the native metadata', async () => {
+		const metadata = { fileSize: 11, mimeType: 'text/plain', fileName: 'file.txt' };
+		azureBlob.getMetadata.mockResolvedValueOnce(metadata);
+
+		expect(await store.getMetadata('a/b/blob.json')).toEqual(metadata);
+	});
+
+	it('should return `null` when the blob is missing', async () => {
+		azureBlob.getMetadata.mockRejectedValueOnce(azureNotFoundError);
+		expect(await store.getMetadata('a/b/blob.json')).toBeNull();
+	});
+});
+
+describe('copy', () => {
+	it('should copy the bytes and preserved metadata to the target key', async () => {
+		azureBlob.get.mockResolvedValueOnce(body as unknown as Readable);
+		const metadata = { fileSize: 11, mimeType: 'text/plain', fileName: 'file.txt' };
+		azureBlob.getMetadata.mockResolvedValueOnce(metadata);
+
+		await store.copy('a/src.bin', 'a/dst.bin');
+
+		expect(azureBlob.get).toHaveBeenCalledWith('a/src.bin', { mode: 'buffer' });
+		expect(azureBlob.put).toHaveBeenCalledWith('a/dst.bin', body, metadata);
+	});
+});
+
+describe('rename', () => {
+	it('should copy to the new key and delete the old one', async () => {
+		azureBlob.get.mockResolvedValueOnce(body as unknown as Readable);
+		azureBlob.getMetadata.mockResolvedValueOnce({ fileSize: 11 });
+
+		await store.rename('a/old.bin', 'a/new.bin');
+
+		expect(azureBlob.put).toHaveBeenCalledWith('a/new.bin', body, { fileSize: 11 });
+		expect(azureBlob.delete).toHaveBeenCalledWith('a/old.bin');
+	});
+
+	it('should be a no-op when old and new keys match', async () => {
+		await store.rename('a/same.bin', 'a/same.bin');
+
+		expect(azureBlob.put).not.toHaveBeenCalled();
+		expect(azureBlob.delete).not.toHaveBeenCalled();
+	});
 });
 
 describe('read', () => {

packages/@n8n/blob-storage/src/__tests__/fs-byte-store.integration.test.ts

@@ -2,6 +2,7 @@ import { UnexpectedError } from 'n8n-workflow';
 import fs, { mkdtemp, rm } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
+import { Readable } from 'node:stream';
 
 import { FsByteStore } from '../fs-byte-store';
 
@@ -126,6 +127,100 @@ describe('delete', () => {
 	});
 });
 
+describe('write (stream)', () => {
+	it('writes a stream atomically and returns the byte count', async () => {
+		const bytes = await store.write('a/streamed.bin', Readable.from(body));
+
+		expect(bytes).toBe(body.length);
+		expect((await store.read('a/streamed.bin'))!.equals(body)).toBe(true);
+		const files = await fs.readdir(join(storagePath, 'a'));
+		expect(files.filter((f) => f.includes('.tmp.'))).toHaveLength(0);
+	});
+});
+
+describe('readStream', () => {
+	it('returns a stream of the stored bytes', async () => {
+		await store.write('a/blob.json', body);
+
+		const stream = await store.readStream('a/blob.json');
+
+		const chunks: Buffer[] = [];
+		for await (const chunk of stream!) chunks.push(Buffer.from(chunk as Buffer));
+		expect(Buffer.concat(chunks).equals(body)).toBe(true);
+	});
+
+	it('returns null for a missing key', async () => {
+		expect(await store.readStream('a/missing.json')).toBeNull();
+	});
+
+	it.each([0, -1, 1.5])('rejects invalid chunkSize %s', async (chunkSize) => {
+		await store.write('a/blob.json', body);
+
+		await expect(store.readStream('a/blob.json', { chunkSize })).rejects.toThrow(UnexpectedError);
+	});
+});
+
+describe('copy', () => {
+	it('copies the bytes to the target key, creating parent dirs', async () => {
+		await store.write('a/src.bin', body);
+
+		await store.copy('a/src.bin', 'b/c/dst.bin');
+
+		expect((await store.read('b/c/dst.bin'))!.equals(body)).toBe(true);
+		expect((await store.read('a/src.bin'))!.equals(body)).toBe(true);
+	});
+});
+
+describe('rename', () => {
+	it('moves the bytes to the new key, creating parent dirs', async () => {
+		await store.write('a/old.bin', body);
+
+		await store.rename('a/old.bin', 'b/new.bin');
+
+		expect(await store.read('a/old.bin')).toBeNull();
+		expect((await store.read('b/new.bin'))!.equals(body)).toBe(true);
+	});
+
+	it('keeps the bytes intact when old and new keys match', async () => {
+		await store.write('a/same.bin', body);
+
+		await store.rename('a/same.bin', 'a/same.bin');
+
+		expect((await store.read('a/same.bin'))!.equals(body)).toBe(true);
+	});
+});
+
+describe('deletePrefix', () => {
+	it('removes everything under the prefix and prunes empty ancestors', async () => {
+		await store.write('wf/exec/binary_data/one.bin', body);
+		await store.write('wf/exec/binary_data/two.bin', body);
+
+		await store.deletePrefix('wf/exec/binary_data');
+
+		expect(await store.read('wf/exec/binary_data/one.bin')).toBeNull();
+		await expect(fs.stat(join(storagePath, 'wf'))).rejects.toThrow();
+	});
+
+	it('leaves siblings outside the prefix intact', async () => {
+		await store.write('wf/exec/binary_data/one.bin', body);
+		await store.write('wf/exec/execution_data/bundle.json', body);
+
+		await store.deletePrefix('wf/exec/binary_data');
+
+		expect(await store.read('wf/exec/execution_data/bundle.json')).not.toBeNull();
+		expect((await fs.stat(join(storagePath, 'wf/exec'))).isDirectory()).toBe(true);
+	});
+
+	it('does not throw for a missing prefix', async () => {
+		await expect(store.deletePrefix('missing/prefix')).resolves.toBeUndefined();
+	});
+
+	it('rejects a prefix that escapes the storage root', async () => {
+		await expect(store.deletePrefix('../escape')).rejects.toThrow(UnexpectedError);
+		await expect(store.deletePrefix('.')).rejects.toThrow(UnexpectedError);
+	});
+});
+
 describe('path traversal guard', () => {
 	it('rejects a key that escapes the storage root', async () => {
 		await expect(store.write('../escape.json', body)).rejects.toThrow(UnexpectedError);

packages/@n8n/blob-storage/src/__tests__/json-store.test.ts

@@ -2,6 +2,8 @@
 /* eslint-disable n8n-local-rules/no-uncaught-json-parse */
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 
+import { Readable } from 'node:stream';
+
 import { JsonStore } from '../json-store';
 import type { ByteStore, ByteStoreKey, StorageLocation } from '../types';
 
@@ -14,7 +16,8 @@ class InMemoryByteStore implements ByteStore {
 	/** Force a non-ENOENT (systemic) failure on read for these keys. */
 	readonly failingKeys = new Set<ByteStoreKey>();
 
-	async write(key: ByteStoreKey, body: Buffer) {
+	async write(key: ByteStoreKey, body: Buffer | Readable) {
+		if (!Buffer.isBuffer(body)) throw new Error('InMemoryByteStore supports Buffer writes only');
 		this.objects.set(key, body);
 		return body.length;
 	}
@@ -24,6 +27,21 @@ class InMemoryByteStore implements ByteStore {
 		return this.objects.get(key) ?? null;
 	}
 
+	async readStream(key: ByteStoreKey): Promise<Readable | null> {
+		const buffer = this.objects.get(key);
+		return buffer ? Readable.from(buffer) : null;
+	}
+
+	async copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey): Promise<void> {
+		const buffer = this.objects.get(sourceKey);
+		if (buffer) this.objects.set(targetKey, buffer);
+	}
+
+	async rename(oldKey: ByteStoreKey, newKey: ByteStoreKey): Promise<void> {
+		await this.copy(oldKey, newKey);
+		this.objects.delete(oldKey);
+	}
+
 	async delete(keys: ByteStoreKey[]): Promise<void> {
 		for (const key of keys) this.objects.delete(key);
 	}

packages/@n8n/blob-storage/src/__tests__/mocks.ts

@@ -18,6 +18,11 @@ export const s3ThrottledError = wrapProviderError(
 	Object.assign(new Error('slow down'), { name: 'ThrottlingException' }),
 );
 
+/** S3 HEAD 404s carry no error code, only a generic `NotFound` name. */
+export const s3HeadNotFoundError = wrapProviderError(
+	Object.assign(new Error('missing'), { name: 'NotFound' }),
+);
+
 export const azureNotFoundError = wrapProviderError(
 	Object.assign(new Error('missing'), { code: 'BlobNotFound' }),
 );

packages/@n8n/blob-storage/src/__tests__/s3-byte-store.test.ts

@@ -1,11 +1,17 @@
 /* eslint-disable @typescript-eslint/unbound-method */
 
-import type { Readable } from 'node:stream';
+import { Readable } from 'node:stream';
 import { mock } from 'vitest-mock-extended';
 
 import type { ObjectStoreService } from '../object-store/object-store.service.ee';
 import { S3ByteStore } from '../s3-byte-store.ee';
-import { body, s3NotFoundError, s3NoSuchBucketError, s3ThrottledError } from './mocks';
+import {
+	body,
+	s3HeadNotFoundError,
+	s3NotFoundError,
+	s3NoSuchBucketError,
+	s3ThrottledError,
+} from './mocks';
 
 let objectStoreService: ReturnType<typeof mock<ObjectStoreService>>;
 let store: S3ByteStore;
@@ -17,7 +23,7 @@ beforeEach(() => {
 
 describe('write', () => {
 	it('should put the bytes at the key and return the byte count', async () => {
-		const bytes = await store.write('a/b/blob.json', body, 'application/json');
+		const bytes = await store.write('a/b/blob.json', body, { mimeType: 'application/json' });
 		expect(objectStoreService.put).toHaveBeenCalledWith('a/b/blob.json', body, {
 			mimeType: 'application/json',
 		});
@@ -30,6 +36,110 @@ describe('write', () => {
 			mimeType: 'application/octet-stream',
 		});
 	});
+
+	it('should forward native metadata', async () => {
+		await store.write('a/b/blob.bin', body, { fileName: 'file.txt', mimeType: 'text/plain' });
+		expect(objectStoreService.put).toHaveBeenCalledWith('a/b/blob.bin', body, {
+			fileName: 'file.txt',
+			mimeType: 'text/plain',
+		});
+	});
+
+	it('should buffer a stream before uploading and return the byte count', async () => {
+		const bytes = await store.write('a/b/blob.bin', Readable.from(body));
+		expect(objectStoreService.put).toHaveBeenCalledWith('a/b/blob.bin', body, {
+			mimeType: 'application/octet-stream',
+		});
+		expect(bytes).toBe(body.length);
+	});
+});
+
+describe('readStream', () => {
+	it('should return the stored stream, forwarding chunkSize', async () => {
+		const stream = Readable.from(body);
+		objectStoreService.get.mockResolvedValueOnce(stream);
+
+		expect(await store.readStream('a/b/blob.json', { chunkSize: 5 })).toBe(stream);
+		expect(objectStoreService.get).toHaveBeenCalledWith('a/b/blob.json', {
+			mode: 'stream',
+			chunkSize: 5,
+		});
+	});
+
+	it('should return `null` when the object is missing', async () => {
+		objectStoreService.get.mockRejectedValueOnce(s3NotFoundError);
+		expect(await store.readStream('a/b/blob.json')).toBeNull();
+	});
+
+	it('should rethrow a systemic error', async () => {
+		objectStoreService.get.mockRejectedValueOnce(s3ThrottledError);
+		await expect(store.readStream('a/b/blob.json')).rejects.toBe(s3ThrottledError);
+	});
+
+	it('should reject a fractional chunkSize before downloading', async () => {
+		await expect(store.readStream('a/b/blob.json', { chunkSize: 1.5 })).rejects.toThrow(
+			'positive integer',
+		);
+		expect(objectStoreService.get).not.toHaveBeenCalled();
+	});
+});
+
+describe('getMetadata', () => {
+	it('should normalize native metadata', async () => {
+		objectStoreService.getMetadata.mockResolvedValueOnce({
+			'content-length': '11',
+			'content-type': 'text/plain',
+			'x-amz-meta-filename': 'file.txt',
+		});
+
+		expect(await store.getMetadata('a/b/blob.json')).toEqual({
+			fileSize: 11,
+			mimeType: 'text/plain',
+			fileName: 'file.txt',
+		});
+	});
+
+	it('should return `null` when the object is missing (HEAD 404)', async () => {
+		objectStoreService.getMetadata.mockRejectedValueOnce(s3HeadNotFoundError);
+		expect(await store.getMetadata('a/b/blob.json')).toBeNull();
+	});
+});
+
+describe('copy', () => {
+	it('should copy the bytes and preserved metadata to the target key', async () => {
+		objectStoreService.get.mockResolvedValueOnce(body as unknown as Readable);
+		objectStoreService.getMetadata.mockResolvedValueOnce({
+			'content-type': 'text/plain',
+			'x-amz-meta-filename': 'file.txt',
+		});
+
+		await store.copy('a/src.bin', 'a/dst.bin');
+
+		expect(objectStoreService.get).toHaveBeenCalledWith('a/src.bin', { mode: 'buffer' });
+		expect(objectStoreService.put).toHaveBeenCalledWith('a/dst.bin', body, {
+			mimeType: 'text/plain',
+			fileName: 'file.txt',
+		});
+	});
+});
+
+describe('rename', () => {
+	it('should copy to the new key and delete the old one', async () => {
+		objectStoreService.get.mockResolvedValueOnce(body as unknown as Readable);
+		objectStoreService.getMetadata.mockResolvedValueOnce({});
+
+		await store.rename('a/old.bin', 'a/new.bin');
+
+		expect(objectStoreService.put).toHaveBeenCalledWith('a/new.bin', body, {});
+		expect(objectStoreService.deleteOne).toHaveBeenCalledWith('a/old.bin');
+	});
+
+	it('should be a no-op when old and new keys match', async () => {
+		await store.rename('a/same.bin', 'a/same.bin');
+
+		expect(objectStoreService.put).not.toHaveBeenCalled();
+		expect(objectStoreService.deleteOne).not.toHaveBeenCalled();
+	});
 });
 
 describe('read', () => {

packages/@n8n/blob-storage/src/__tests__/stream-utils.test.ts

@@ -52,7 +52,7 @@ describe('stream-utils', () => {
 			expect(chunks).toEqual([]);
 		});
 
-		it.each([0, -1])('should throw when chunkSize is %s', (chunkSize) => {
+		it.each([0, -1, 1.5])('should throw when chunkSize is %s', (chunkSize) => {
 			expect(() => createFixedSizeChunker(chunkSize)).toThrow(UnexpectedError);
 		});
 	});

packages/@n8n/blob-storage/src/azure-byte-store.ee.ts

@@ -1,17 +1,24 @@
+import { binaryToBuffer } from '@n8n/backend-network';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 import chunk from 'lodash/chunk';
+import type { Readable } from 'node:stream';
 
 import type { AzureBlobService } from './azure-blob/azure-blob.service.ee';
-import type { ByteStore, ByteStoreKey } from './types';
+import { assertChunkSize } from './stream-utils';
+import type { BlobMetadata, ByteStore, ByteStoreKey, PreWriteBlobMetadata } from './types';
 
 const MAX_DELETE_CONCURRENCY = 50;
 
 export class AzureByteStore implements ByteStore {
 	constructor(private readonly azureBlob: AzureBlobService) {}
 
-	async write(key: ByteStoreKey, body: Buffer, contentType?: string) {
-		await this.azureBlob.put(key, body, { mimeType: contentType ?? 'application/octet-stream' });
-		return body.length;
+	async write(key: ByteStoreKey, body: Buffer | Readable, metadata: PreWriteBlobMetadata = {}) {
+		const buffer = Buffer.isBuffer(body) ? body : await binaryToBuffer(body);
+		await this.azureBlob.put(key, buffer, {
+			...metadata,
+			mimeType: metadata.mimeType ?? 'application/octet-stream',
+		});
+		return buffer.length;
 	}
 
 	async read(key: ByteStoreKey): Promise<Buffer | null> {
@@ -23,12 +30,45 @@ export class AzureByteStore implements ByteStore {
 		}
 	}
 
+	async readStream(key: ByteStoreKey, { chunkSize }: { chunkSize?: number } = {}) {
+		if (chunkSize !== undefined) assertChunkSize(chunkSize);
+		try {
+			return await this.azureBlob.get(key, { mode: 'stream', chunkSize });
+		} catch (error) {
+			if (this.isNotFound(error)) return null;
+			throw error;
+		}
+	}
+
+	async copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey) {
+		const buffer = await this.azureBlob.get(sourceKey, { mode: 'buffer' });
+		const metadata = await this.azureBlob.getMetadata(sourceKey);
+		await this.azureBlob.put(targetKey, buffer, metadata);
+	}
+
+	async rename(oldKey: ByteStoreKey, newKey: ByteStoreKey) {
+		if (oldKey === newKey) return;
+		await this.copy(oldKey, newKey);
+		await this.azureBlob.delete(oldKey);
+	}
+
 	async delete(keys: ByteStoreKey[]): Promise<void> {
 		for (const batch of chunk(keys, MAX_DELETE_CONCURRENCY)) {
 			await Promise.all(batch.map(async (key) => await this.azureBlob.delete(key)));
 		}
 	}
 
+	async getMetadata(key: ByteStoreKey): Promise<BlobMetadata | null> {
+		try {
+			return await this.azureBlob.getMetadata(key);
+		} catch (error) {
+			if (this.isNotFound(error)) return null;
+			throw error;
+		}
+	}
+
+	// private methods
+
 	private isNotFound(error: unknown) {
 		const original = ensureError(error).cause ?? error;
 		if (typeof original !== 'object' || original === null) return false;

packages/@n8n/blob-storage/src/fs-byte-store.ts

@@ -3,7 +3,9 @@ import { UnexpectedError } from 'n8n-workflow';
 import { randomUUID } from 'node:crypto';
 import fs from 'node:fs/promises';
 import path from 'node:path';
+import type { Readable } from 'node:stream';
 
+import { assertChunkSize } from './stream-utils';
 import type { ByteStore, ByteStoreKey } from './types';
 
 export type FsByteStoreOptions = {
@@ -21,7 +23,7 @@ export class FsByteStore implements ByteStore {
 		await assertDir(this.options.storagePath);
 	}
 
-	async write(key: ByteStoreKey, body: Buffer) {
+	async write(key: ByteStoreKey, body: Buffer | Readable) {
 		const writePath = this.getAbsolutePath(key);
 		await assertDir(path.dirname(writePath));
 		const tempPath = `${writePath}.tmp.${process.pid}.${randomUUID()}`;
@@ -30,9 +32,10 @@ export class FsByteStore implements ByteStore {
 
 		try {
 			await fs.writeFile(tempPath, body);
+			const bytesWritten = Buffer.isBuffer(body) ? body.length : (await fs.stat(tempPath)).size;
 			await fs.rename(tempPath, writePath);
 			success = true;
-			return body.length;
+			return bytesWritten;
 		} finally {
 			if (!success) {
 				await fs.rm(tempPath, { force: true }).catch((error) => this.options.reportError(error));
@@ -50,6 +53,43 @@ export class FsByteStore implements ByteStore {
 		}
 	}
 
+	async readStream(key: ByteStoreKey, { chunkSize }: { chunkSize?: number } = {}) {
+		if (chunkSize !== undefined) assertChunkSize(chunkSize);
+		const readPath = this.getAbsolutePath(key);
+		let fileHandle;
+		try {
+			fileHandle = await fs.open(readPath, 'r');
+		} catch (error) {
+			if (this.isFileNotFound(error)) return null;
+			throw error;
+		}
+		try {
+			return fileHandle.createReadStream(chunkSize ? { highWaterMark: chunkSize } : {});
+		} catch (error) {
+			await fileHandle.close().catch(() => {});
+			throw error;
+		}
+	}
+
+	async copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey) {
+		const targetPath = this.getAbsolutePath(targetKey);
+		await assertDir(path.dirname(targetPath));
+		await fs.copyFile(this.getAbsolutePath(sourceKey), targetPath);
+	}
+
+	async rename(oldKey: ByteStoreKey, newKey: ByteStoreKey) {
+		if (oldKey === newKey) return;
+		const newPath = this.getAbsolutePath(newKey);
+		await assertDir(path.dirname(newPath));
+		await fs.rename(this.getAbsolutePath(oldKey), newPath);
+	}
+
+	async deletePrefix(prefix: string) {
+		const dir = this.getAbsolutePath(prefix);
+		await fs.rm(dir, { recursive: true, force: true });
+		await this.removeEmptyAncestors(path.dirname(dir));
+	}
+
 	async delete(keys: ByteStoreKey[]) {
 		const deletePaths = keys.map((key) => this.getAbsolutePath(key));
 		await Promise.all(deletePaths.map(async (p) => await fs.rm(p, { force: true })));

packages/@n8n/blob-storage/src/json-store.ts

@@ -39,7 +39,7 @@ export class JsonStore<Ref, Payload extends object> {
 				jsonStringify({ ...payload, version: this.options.version }),
 				'utf-8',
 			);
-			return await store.write(this.options.key(ref), body, 'application/json');
+			return await store.write(this.options.key(ref), body, { mimeType: 'application/json' });
 		} catch (error) {
 			throw this.options.createWriteError(ref, error);
 		}

packages/@n8n/blob-storage/src/s3-byte-store.ee.ts

@@ -1,36 +1,97 @@
+import { binaryToBuffer } from '@n8n/backend-network';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
+import type { Readable } from 'node:stream';
 
 import type { ObjectStoreService } from './object-store/object-store.service.ee';
-import type { ByteStore, ByteStoreKey } from './types';
+import { assertChunkSize } from './stream-utils';
+import type { BlobMetadata, ByteStore, ByteStoreKey, PreWriteBlobMetadata } from './types';
 
 export class S3ByteStore implements ByteStore {
 	constructor(private readonly objectStore: ObjectStoreService) {}
 
-	async write(key: ByteStoreKey, body: Buffer, contentType?: string) {
-		await this.objectStore.put(key, body, { mimeType: contentType ?? 'application/octet-stream' });
-		return body.length;
+	async write(key: ByteStoreKey, body: Buffer | Readable, metadata: PreWriteBlobMetadata = {}) {
+		const buffer = Buffer.isBuffer(body) ? body : await binaryToBuffer(body);
+		await this.objectStore.put(key, buffer, {
+			...metadata,
+			mimeType: metadata.mimeType ?? 'application/octet-stream',
+		});
+		return buffer.length;
 	}
 
 	async read(key: ByteStoreKey) {
 		try {
 			return await this.objectStore.get(key, { mode: 'buffer' });
 		} catch (error) {
-			if (this.isNotFound(error)) return null;
+			if (this.isNoSuchKey(error)) return null;
 			throw error;
 		}
 	}
 
+	async readStream(key: ByteStoreKey, { chunkSize }: { chunkSize?: number } = {}) {
+		if (chunkSize !== undefined) assertChunkSize(chunkSize);
+		try {
+			return await this.objectStore.get(key, { mode: 'stream', chunkSize });
+		} catch (error) {
+			if (this.isNoSuchKey(error)) return null;
+			throw error;
+		}
+	}
+
+	async copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey) {
+		const buffer = await this.objectStore.get(sourceKey, { mode: 'buffer' });
+		await this.objectStore.put(targetKey, buffer, await this.readNativeMetadata(sourceKey));
+	}
+
+	async rename(oldKey: ByteStoreKey, newKey: ByteStoreKey) {
+		if (oldKey === newKey) return;
+		await this.copy(oldKey, newKey);
+		await this.objectStore.deleteOne(oldKey);
+	}
+
 	async delete(keys: ByteStoreKey[]): Promise<void> {
 		if (keys.length === 0) return;
 		await this.objectStore.deleteByKeys(keys);
 	}
 
+	async getMetadata(key: ByteStoreKey): Promise<BlobMetadata | null> {
+		try {
+			const headers = await this.objectStore.getMetadata(key);
+			const metadata: BlobMetadata = { fileSize: Number(headers['content-length'] ?? 0) };
+			if (headers['content-type']) metadata.mimeType = headers['content-type'];
+			if (headers['x-amz-meta-filename']) metadata.fileName = headers['x-amz-meta-filename'];
+			return metadata;
+		} catch (error) {
+			if (this.isHeadNotFound(error)) return null;
+			throw error;
+		}
+	}
+
 	// private methods
 
-	private isNotFound(error: unknown) {
+	private async readNativeMetadata(key: ByteStoreKey): Promise<PreWriteBlobMetadata> {
+		const headers = await this.objectStore.getMetadata(key);
+		const metadata: PreWriteBlobMetadata = {};
+		if (headers['content-type']) metadata.mimeType = headers['content-type'];
+		if (headers['x-amz-meta-filename']) metadata.fileName = headers['x-amz-meta-filename'];
+		return metadata;
+	}
+
+	/** Missing object on a GET. Bucket-level failures surface as `NoSuchBucket` and rethrow. */
+	private isNoSuchKey(error: unknown) {
+		return this.errorName(error) === 'NoSuchKey';
+	}
+
+	/**
+	 * Missing object on a HEAD, which carries no error code, so a 404 cannot be told apart
+	 * from a missing bucket. Acceptable: bucket existence is verified at startup (`init`).
+	 */
+	private isHeadNotFound(error: unknown) {
+		return this.errorName(error) === 'NotFound';
+	}
+
+	private errorName(error: unknown) {
 		const original = ensureError(error).cause ?? error;
-		if (typeof original !== 'object' || original === null) return false;
-		const name = 'name' in original ? original.name : undefined;
-		return name === 'NoSuchKey' || name === 'NotFound';
+		if (typeof original !== 'object' || original === null) return undefined;
+		return 'name' in original ? original.name : undefined;
 	}
 }

packages/@n8n/blob-storage/src/stream-utils.ts

@@ -1,6 +1,13 @@
 import { UnexpectedError } from 'n8n-workflow';
 import { Transform } from 'node:stream';
 
+/** Throws unless `chunkSize` is a positive integer. */
+export function assertChunkSize(chunkSize: number): void {
+	if (!Number.isInteger(chunkSize) || chunkSize <= 0) {
+		throw new UnexpectedError(`chunkSize must be a positive integer, got ${chunkSize}`);
+	}
+}
+
 /**
  * A `Transform` that re-emits its input as chunks of exactly `chunkSize` bytes, with a possibly smaller final chunk.
  * `chunkSize` must be a positive integer: values `<= 0` throws an `UnexpectedError`.
@@ -14,9 +21,7 @@ import { Transform } from 'node:stream';
  * `.pipe()` does neither.
  */
 export function createFixedSizeChunker(chunkSize: number): Transform {
-	if (chunkSize <= 0) {
-		throw new UnexpectedError(`createFixedSizeChunker requires chunkSize > 0, got ${chunkSize}`);
-	}
+	assertChunkSize(chunkSize);
 
 	const queue: Buffer[] = [];
 	let queued = 0;

packages/@n8n/blob-storage/src/types.ts

@@ -1,3 +1,5 @@
+import type { Readable } from 'node:stream';
+
 /** Metadata stored alongside a blob. */
 export type BlobMetadata = {
 	fileName?: string;
@@ -16,13 +18,32 @@ export type ByteStoreKey = string;
 export interface ByteStore {
 	init?(): Promise<void>;
 
-	/** Returns the number of bytes written. */
-	write(key: ByteStoreKey, body: Buffer, contentType?: string): Promise<number>;
+	/** Returns the number of bytes written. `metadata` is stored natively where supported, `fs` ignores it. */
+	write(
+		key: ByteStoreKey,
+		body: Buffer | Readable,
+		metadata?: PreWriteBlobMetadata,
+	): Promise<number>;
 
 	/** Returns `null` when no object exists for `key`. */
 	read(key: ByteStoreKey): Promise<Buffer | null>;
 
+	/** Returns `null` when no object exists. `chunkSize` (positive integer) fixes the emitted chunk size (final chunk may be smaller). */
+	readStream(key: ByteStoreKey, opts?: { chunkSize?: number }): Promise<Readable | null>;
+
+	/** Preserves native metadata where supported. */
+	copy(sourceKey: ByteStoreKey, targetKey: ByteStoreKey): Promise<void>;
+
+	/** Preserves native metadata where supported. Same-key rename is a no-op, even for a missing source. */
+	rename(oldKey: ByteStoreKey, newKey: ByteStoreKey): Promise<void>;
+
 	delete(keys: ByteStoreKey[]): Promise<void>;
+
+	/** `null` when no object exists. Absent on `fs`, which keeps sidecar entries at the domain layer. */
+	getMetadata?(key: ByteStoreKey): Promise<BlobMetadata | null>;
+
+	/** Recursively deletes under `prefix`. Absent where the backend can't list or delegates bulk deletion (e.g. lifecycle policies). */
+	deletePrefix?(prefix: string): Promise<void>;
 }
 
 /** A stored JSON entry. */

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

@@ -33,7 +33,7 @@ it('should write the version-1 bundle at the stable execution-data key', async (
 	expect(fsByteStore.write).toHaveBeenCalledWith(
 		`workflows/${workflowId}/executions/${executionId}/execution_data/bundle.json`,
 		expect.any(Buffer),
-		'application/json',
+		{ mimeType: 'application/json' },
 	);
 	const [_, body] = fsByteStore.write.mock.calls[0];
 	expect(JSON.parse(body.toString('utf-8'))).toEqual({ ...payload, version: 1 });