✨ Introduce binary data management (#2059)

* introduce binary data management

* merge fixes

* fixes

* init binary data manager for other modes

* improve binary manager

* improve binary manager

* delete binary data on executions delete

* lazy delete non-saved executions binary data

* merge fixes + error handing

* improve structure

* leftovers and cleanups

* formatting

* fix config description

* fixes

* fix races

* duplicate binary data for execute workflow node

* clean up and cr

* update mode name, add binary mode to diagnostics

* update mode name, add prefix to filename

* update filename

* allow multiple modes, backward compatibility

* improve file and id naming

* use execution id for binary data storage

* delete binary data by execution id

* add meta for persisted binary data

* delete marked persisted files

* mark deletion by executionid

* add env var for persisted binary data ttl

* improvements

* lint fix

* fix env var description

* cleanup

* cleanup

* ⚡ Minor improvements

Co-authored-by: Jan Oberhauser <jan.oberhauser@gmail.com>

packages/core/src/BinaryDataManager/FileSystem.ts

@@ -0,0 +1,214 @@
+import { promises as fs } from 'fs';
+import * as path from 'path';
+import { v4 as uuid } from 'uuid';
+
+import { IBinaryDataConfig, IBinaryDataManager } from '../Interfaces';
+
+const PREFIX_METAFILE = 'binarymeta';
+const PREFIX_PERSISTED_METAFILE = 'persistedmeta';
+
+export class BinaryDataFileSystem implements IBinaryDataManager {
+	private storagePath: string;
+
+	private binaryDataTTL: number;
+
+	private persistedBinaryDataTTL: number;
+
+	constructor(config: IBinaryDataConfig) {
+		this.storagePath = config.localStoragePath;
+		this.binaryDataTTL = config.binaryDataTTL;
+		this.persistedBinaryDataTTL = config.persistedBinaryDataTTL;
+	}
+
+	async init(startPurger = false): Promise<void> {
+		if (startPurger) {
+			setInterval(async () => {
+				await this.deleteMarkedFiles();
+			}, this.binaryDataTTL * 30000);
+
+			setInterval(async () => {
+				await this.deleteMarkedPersistedFiles();
+			}, this.persistedBinaryDataTTL * 30000);
+		}
+
+		return fs
+			.readdir(this.storagePath)
+			.catch(async () => fs.mkdir(this.storagePath, { recursive: true }))
+			.then(async () => fs.readdir(this.getBinaryDataMetaPath()))
+			.catch(async () => fs.mkdir(this.getBinaryDataMetaPath(), { recursive: true }))
+			.then(async () => fs.readdir(this.getBinaryDataPersistMetaPath()))
+			.catch(async () => fs.mkdir(this.getBinaryDataPersistMetaPath(), { recursive: true }))
+			.then(async () => this.deleteMarkedFiles())
+			.then(async () => this.deleteMarkedPersistedFiles())
+			.then(() => {});
+	}
+
+	async storeBinaryData(binaryBuffer: Buffer, executionId: string): Promise<string> {
+		const binaryDataId = this.generateFileName(executionId);
+		return this.addBinaryIdToPersistMeta(executionId, binaryDataId).then(async () =>
+			this.saveToLocalStorage(binaryBuffer, binaryDataId).then(() => binaryDataId),
+		);
+	}
+
+	async retrieveBinaryDataByIdentifier(identifier: string): Promise<Buffer> {
+		return this.retrieveFromLocalStorage(identifier);
+	}
+
+	async markDataForDeletionByExecutionId(executionId: string): Promise<void> {
+		const tt = new Date(new Date().getTime() + this.binaryDataTTL * 60000);
+		return fs.writeFile(
+			path.join(this.getBinaryDataMetaPath(), `${PREFIX_METAFILE}_${executionId}_${tt.valueOf()}`),
+			'',
+		);
+	}
+
+	async deleteMarkedFiles(): Promise<void> {
+		return this.deleteMarkedFilesByMeta(this.getBinaryDataMetaPath(), PREFIX_METAFILE);
+	}
+
+	async deleteMarkedPersistedFiles(): Promise<void> {
+		return this.deleteMarkedFilesByMeta(
+			this.getBinaryDataPersistMetaPath(),
+			PREFIX_PERSISTED_METAFILE,
+		);
+	}
+
+	private async addBinaryIdToPersistMeta(executionId: string, identifier: string): Promise<void> {
+		const currentTime = new Date().getTime();
+		const timeAtNextHour = currentTime + 3600000 - (currentTime % 3600000);
+		const timeoutTime = timeAtNextHour + this.persistedBinaryDataTTL * 60000;
+
+		const filePath = path.join(
+			this.getBinaryDataPersistMetaPath(),
+			`${PREFIX_PERSISTED_METAFILE}_${executionId}_${timeoutTime}`,
+		);
+
+		return fs
+			.readFile(filePath)
+			.catch(async () => fs.writeFile(filePath, identifier))
+			.then(() => {});
+	}
+
+	private async deleteMarkedFilesByMeta(metaPath: string, filePrefix: string): Promise<void> {
+		const currentTimeValue = new Date().valueOf();
+		const metaFileNames = await fs.readdir(metaPath);
+
+		const execsAdded: { [key: string]: number } = {};
+
+		const proms = metaFileNames.reduce(
+			(prev, curr) => {
+				const [prefix, executionId, ts] = curr.split('_');
+
+				if (prefix !== filePrefix) {
+					return prev;
+				}
+
+				const execTimestamp = parseInt(ts, 10);
+
+				if (execTimestamp < currentTimeValue) {
+					if (execsAdded[executionId]) {
+						// do not delete data, only meta file
+						prev.push(this.deleteMetaFileByPath(path.join(metaPath, curr)));
+						return prev;
+					}
+
+					execsAdded[executionId] = 1;
+					prev.push(
+						this.deleteBinaryDataByExecutionId(executionId).then(async () =>
+							this.deleteMetaFileByPath(path.join(metaPath, curr)),
+						),
+					);
+				}
+
+				return prev;
+			},
+			[Promise.resolve()],
+		);
+
+		return Promise.all(proms).then(() => {});
+	}
+
+	async duplicateBinaryDataByIdentifier(binaryDataId: string, prefix: string): Promise<string> {
+		const newBinaryDataId = this.generateFileName(prefix);
+
+		return fs
+			.copyFile(
+				path.join(this.storagePath, binaryDataId),
+				path.join(this.storagePath, newBinaryDataId),
+			)
+			.then(() => newBinaryDataId);
+	}
+
+	async deleteBinaryDataByExecutionId(executionId: string): Promise<void> {
+		const regex = new RegExp(`${executionId}_*`);
+		const filenames = await fs.readdir(path.join(this.storagePath));
+
+		const proms = filenames.reduce(
+			(allProms, filename) => {
+				if (regex.test(filename)) {
+					allProms.push(fs.rm(path.join(this.storagePath, filename)));
+				}
+
+				return allProms;
+			},
+			[Promise.resolve()],
+		);
+
+		return Promise.all(proms).then(async () => Promise.resolve());
+	}
+
+	async deleteBinaryDataByIdentifier(identifier: string): Promise<void> {
+		return this.deleteFromLocalStorage(identifier);
+	}
+
+	async persistBinaryDataForExecutionId(executionId: string): Promise<void> {
+		return fs.readdir(this.getBinaryDataPersistMetaPath()).then(async (metafiles) => {
+			const proms = metafiles.reduce(
+				(prev, curr) => {
+					if (curr.startsWith(`${PREFIX_PERSISTED_METAFILE}_${executionId}_`)) {
+						prev.push(fs.rm(path.join(this.getBinaryDataPersistMetaPath(), curr)));
+						return prev;
+					}
+
+					return prev;
+				},
+				[Promise.resolve()],
+			);
+
+			return Promise.all(proms).then(() => {});
+		});
+	}
+
+	private generateFileName(prefix: string): string {
+		return `${prefix}_${uuid()}`;
+	}
+
+	private getBinaryDataMetaPath() {
+		return path.join(this.storagePath, 'meta');
+	}
+
+	private getBinaryDataPersistMetaPath() {
+		return path.join(this.storagePath, 'persistMeta');
+	}
+
+	private async deleteMetaFileByPath(metafilePath: string): Promise<void> {
+		return fs.rm(metafilePath);
+	}
+
+	private async deleteFromLocalStorage(identifier: string) {
+		return fs.rm(path.join(this.storagePath, identifier));
+	}
+
+	private async saveToLocalStorage(data: Buffer, identifier: string) {
+		await fs.writeFile(path.join(this.storagePath, identifier), data);
+	}
+
+	private async retrieveFromLocalStorage(identifier: string): Promise<Buffer> {
+		const filePath = path.join(this.storagePath, identifier);
+		try {
+			return await fs.readFile(filePath);
+		} catch (e) {
+			throw new Error(`Error finding file: ${filePath}`);
+		}
+	}
+}

packages/core/src/BinaryDataManager/index.ts

@@ -0,0 +1,187 @@
+import { IBinaryData, INodeExecutionData } from 'n8n-workflow';
+import { BINARY_ENCODING } from '../Constants';
+import { IBinaryDataConfig, IBinaryDataManager } from '../Interfaces';
+import { BinaryDataFileSystem } from './FileSystem';
+
+export class BinaryDataManager {
+	private static instance: BinaryDataManager;
+
+	private managers: {
+		[key: string]: IBinaryDataManager;
+	};
+
+	private binaryDataMode: string;
+
+	private availableModes: string[];
+
+	constructor(config: IBinaryDataConfig) {
+		this.binaryDataMode = config.mode;
+		this.availableModes = config.availableModes.split(',');
+		this.managers = {};
+	}
+
+	static async init(config: IBinaryDataConfig, mainManager = false): Promise<void> {
+		if (BinaryDataManager.instance) {
+			throw new Error('Binary Data Manager already initialized');
+		}
+
+		BinaryDataManager.instance = new BinaryDataManager(config);
+
+		if (BinaryDataManager.instance.availableModes.includes('filesystem')) {
+			BinaryDataManager.instance.managers.filesystem = new BinaryDataFileSystem(config);
+			await BinaryDataManager.instance.managers.filesystem.init(mainManager);
+		}
+
+		return undefined;
+	}
+
+	static getInstance(): BinaryDataManager {
+		if (!BinaryDataManager.instance) {
+			throw new Error('Binary Data Manager not initialized');
+		}
+
+		return BinaryDataManager.instance;
+	}
+
+	async storeBinaryData(
+		binaryData: IBinaryData,
+		binaryBuffer: Buffer,
+		executionId: string,
+	): Promise<IBinaryData> {
+		const retBinaryData = binaryData;
+
+		if (this.managers[this.binaryDataMode]) {
+			return this.managers[this.binaryDataMode]
+				.storeBinaryData(binaryBuffer, executionId)
+				.then((filename) => {
+					retBinaryData.id = this.generateBinaryId(filename);
+					return retBinaryData;
+				});
+		}
+
+		retBinaryData.data = binaryBuffer.toString(BINARY_ENCODING);
+		return binaryData;
+	}
+
+	async retrieveBinaryData(binaryData: IBinaryData): Promise<Buffer> {
+		if (binaryData.id) {
+			return this.retrieveBinaryDataByIdentifier(binaryData.id);
+		}
+
+		return Buffer.from(binaryData.data, BINARY_ENCODING);
+	}
+
+	async retrieveBinaryDataByIdentifier(identifier: string): Promise<Buffer> {
+		const { mode, id } = this.splitBinaryModeFileId(identifier);
+		if (this.managers[mode]) {
+			return this.managers[mode].retrieveBinaryDataByIdentifier(id);
+		}
+
+		throw new Error('Storage mode used to store binary data not available');
+	}
+
+	async markDataForDeletionByExecutionId(executionId: string): Promise<void> {
+		if (this.managers[this.binaryDataMode]) {
+			return this.managers[this.binaryDataMode].markDataForDeletionByExecutionId(executionId);
+		}
+
+		return Promise.resolve();
+	}
+
+	async persistBinaryDataForExecutionId(executionId: string): Promise<void> {
+		if (this.managers[this.binaryDataMode]) {
+			return this.managers[this.binaryDataMode].persistBinaryDataForExecutionId(executionId);
+		}
+
+		return Promise.resolve();
+	}
+
+	async deleteBinaryDataByExecutionId(executionId: string): Promise<void> {
+		if (this.managers[this.binaryDataMode]) {
+			return this.managers[this.binaryDataMode].deleteBinaryDataByExecutionId(executionId);
+		}
+
+		return Promise.resolve();
+	}
+
+	async duplicateBinaryData(
+		inputData: Array<INodeExecutionData[] | null> | unknown,
+		executionId: string,
+	): Promise<INodeExecutionData[][]> {
+		if (inputData && this.managers[this.binaryDataMode]) {
+			const returnInputData = (inputData as INodeExecutionData[][]).map(
+				async (executionDataArray) => {
+					if (executionDataArray) {
+						return Promise.all(
+							executionDataArray.map((executionData) => {
+								if (executionData.binary) {
+									return this.duplicateBinaryDataInExecData(executionData, executionId);
+								}
+
+								return executionData;
+							}),
+						);
+					}
+
+					return executionDataArray;
+				},
+			);
+
+			return Promise.all(returnInputData);
+		}
+
+		return Promise.resolve(inputData as INodeExecutionData[][]);
+	}
+
+	private generateBinaryId(filename: string) {
+		return `${this.binaryDataMode}:${filename}`;
+	}
+
+	private splitBinaryModeFileId(fileId: string): { mode: string; id: string } {
+		const [mode, id] = fileId.split(':');
+		return { mode, id };
+	}
+
+	private async duplicateBinaryDataInExecData(
+		executionData: INodeExecutionData,
+		executionId: string,
+	): Promise<INodeExecutionData> {
+		const binaryManager = this.managers[this.binaryDataMode];
+
+		if (executionData.binary) {
+			const binaryDataKeys = Object.keys(executionData.binary);
+			const bdPromises = binaryDataKeys.map(async (key: string) => {
+				if (!executionData.binary) {
+					return { key, newId: undefined };
+				}
+
+				const binaryDataId = executionData.binary[key].id;
+				if (!binaryDataId) {
+					return { key, newId: undefined };
+				}
+
+				return binaryManager
+					?.duplicateBinaryDataByIdentifier(
+						this.splitBinaryModeFileId(binaryDataId).id,
+						executionId,
+					)
+					.then((filename) => ({
+						newId: this.generateBinaryId(filename),
+						key,
+					}));
+			});
+
+			return Promise.all(bdPromises).then((b) => {
+				return b.reduce((acc, curr) => {
+					if (acc.binary && curr) {
+						acc.binary[curr.key].id = curr.newId;
+					}
+
+					return acc;
+				}, executionData);
+			});
+		}
+
+		return executionData;
+	}
+}

packages/core/src/Interfaces.ts

@@ -234,3 +234,23 @@ export interface IWorkflowData {
 	pollResponses?: IPollResponse[];
 	triggerResponses?: ITriggerResponse[];
 }
+
+export interface IBinaryDataConfig {
+	mode: 'default' | 'filesystem';
+	availableModes: string;
+	localStoragePath: string;
+	binaryDataTTL: number;
+	persistedBinaryDataTTL: number;
+}
+
+export interface IBinaryDataManager {
+	init(startPurger: boolean): Promise<void>;
+	storeBinaryData(binaryBuffer: Buffer, executionId: string): Promise<string>;
+	retrieveBinaryDataByIdentifier(identifier: string): Promise<Buffer>;
+	markDataForDeletionByExecutionId(executionId: string): Promise<void>;
+	deleteMarkedFiles(): Promise<unknown>;
+	deleteBinaryDataByIdentifier(identifier: string): Promise<void>;
+	duplicateBinaryDataByIdentifier(binaryDataId: string, prefix: string): Promise<string>;
+	deleteBinaryDataByExecutionId(executionId: string): Promise<void>;
+	persistBinaryDataForExecutionId(executionId: string): Promise<void>;
+}

packages/core/src/NodeExecuteFunctions.ts

@@ -73,9 +73,9 @@ import { lookup } from 'mime-types';
 
 import axios, { AxiosProxyConfig, AxiosRequestConfig, Method } from 'axios';
 import { URL, URLSearchParams } from 'url';
+import { BinaryDataManager } from './BinaryDataManager';
 // eslint-disable-next-line import/no-cycle
 import {
-	BINARY_ENCODING,
 	ICredentialTestFunctions,
 	IHookFunctions,
 	ILoadOptionsFunctions,
@@ -682,7 +682,7 @@ export async function getBinaryDataBuffer(
 	inputIndex: number,
 ): Promise<Buffer> {
 	const binaryData = inputData.main![inputIndex]![itemIndex]!.binary![propertyName]!;
-	return Buffer.from(binaryData.data, BINARY_ENCODING);
+	return BinaryDataManager.getInstance().retrieveBinaryData(binaryData);
 }
 
 /**
@@ -697,6 +697,7 @@ export async function getBinaryDataBuffer(
  */
 export async function prepareBinaryData(
 	binaryData: Buffer,
+	executionId: string,
 	filePath?: string,
 	mimeType?: string,
 ): Promise<IBinaryData> {
@@ -727,10 +728,7 @@ export async function prepareBinaryData(
 
 	const returnData: IBinaryData = {
 		mimeType,
-		// TODO: Should program it in a way that it does not have to converted to base64
-		//       It should only convert to and from base64 when saved in database because
-		//       of for example an error or when there is a wait node.
-		data: binaryData.toString(BINARY_ENCODING),
+		data: '',
 	};
 
 	if (filePath) {
@@ -753,7 +751,7 @@ export async function prepareBinaryData(
 		}
 	}
 
-	return returnData;
+	return BinaryDataManager.getInstance().storeBinaryData(returnData, binaryData, executionId);
 }
 
 /**
@@ -1370,7 +1368,19 @@ export function getExecutePollFunctions(
 			},
 			helpers: {
 				httpRequest,
-				prepareBinaryData,
+				async prepareBinaryData(
+					binaryData: Buffer,
+					filePath?: string,
+					mimeType?: string,
+				): Promise<IBinaryData> {
+					return prepareBinaryData.call(
+						this,
+						binaryData,
+						additionalData.executionId!,
+						filePath,
+						mimeType,
+					);
+				},
 				request: proxyRequestToAxios,
 				async requestOAuth2(
 					this: IAllExecuteFunctions,
@@ -1476,8 +1486,19 @@ export function getExecuteTriggerFunctions(
 			},
 			helpers: {
 				httpRequest,
-				prepareBinaryData,
-
+				async prepareBinaryData(
+					binaryData: Buffer,
+					filePath?: string,
+					mimeType?: string,
+				): Promise<IBinaryData> {
+					return prepareBinaryData.call(
+						this,
+						binaryData,
+						additionalData.executionId!,
+						filePath,
+						mimeType,
+					);
+				},
 				request: proxyRequestToAxios,
 				async requestOAuth2(
 					this: IAllExecuteFunctions,
@@ -1553,7 +1574,14 @@ export function getExecuteFunctions(
 				workflowInfo: IExecuteWorkflowInfo,
 				inputData?: INodeExecutionData[],
 			): Promise<any> {
-				return additionalData.executeWorkflow(workflowInfo, additionalData, inputData);
+				return additionalData
+					.executeWorkflow(workflowInfo, additionalData, inputData)
+					.then(async (result) =>
+						BinaryDataManager.getInstance().duplicateBinaryData(
+							result,
+							additionalData.executionId!,
+						),
+					);
 			},
 			getContext(type: string): IContextObject {
 				return NodeHelpers.getContext(runExecutionData, type, node);
@@ -1672,7 +1700,19 @@ export function getExecuteFunctions(
 			},
 			helpers: {
 				httpRequest,
-				prepareBinaryData,
+				async prepareBinaryData(
+					binaryData: Buffer,
+					filePath?: string,
+					mimeType?: string,
+				): Promise<IBinaryData> {
+					return prepareBinaryData.call(
+						this,
+						binaryData,
+						additionalData.executionId!,
+						filePath,
+						mimeType,
+					);
+				},
 				async getBinaryDataBuffer(
 					itemIndex: number,
 					propertyName: string,
@@ -1853,7 +1893,19 @@ export function getExecuteSingleFunctions(
 			},
 			helpers: {
 				httpRequest,
-				prepareBinaryData,
+				async prepareBinaryData(
+					binaryData: Buffer,
+					filePath?: string,
+					mimeType?: string,
+				): Promise<IBinaryData> {
+					return prepareBinaryData.call(
+						this,
+						binaryData,
+						additionalData.executionId!,
+						filePath,
+						mimeType,
+					);
+				},
 				request: proxyRequestToAxios,
 				async requestOAuth2(
 					this: IAllExecuteFunctions,
@@ -2234,7 +2286,19 @@ export function getExecuteWebhookFunctions(
 			prepareOutputData: NodeHelpers.prepareOutputData,
 			helpers: {
 				httpRequest,
-				prepareBinaryData,
+				async prepareBinaryData(
+					binaryData: Buffer,
+					filePath?: string,
+					mimeType?: string,
+				): Promise<IBinaryData> {
+					return prepareBinaryData.call(
+						this,
+						binaryData,
+						additionalData.executionId!,
+						filePath,
+						mimeType,
+					);
+				},
 				request: proxyRequestToAxios,
 				async requestOAuth2(
 					this: IAllExecuteFunctions,

packages/core/src/index.ts

@@ -10,6 +10,7 @@ try {
 
 export * from './ActiveWorkflows';
 export * from './ActiveWebhooks';
+export * from './BinaryDataManager';
 export * from './Constants';
 export * from './Credentials';
 export * from './Interfaces';