refactor(core): Simplify postExecutePromises

packages/cli/src/__tests__/active-executions.test.ts

@@ -96,7 +96,7 @@ describe('ActiveExecutions', () => {
 	test('Should remove an existing execution', async () => {
 		const newExecution = mockExecutionData();
 		const executionId = await activeExecutions.add(newExecution);
-		activeExecutions.remove(executionId);
+		activeExecutions.finishExecution(executionId);
 
 		expect(activeExecutions.getActiveExecutions().length).toBe(0);
 	});
@@ -110,7 +110,7 @@ describe('ActiveExecutions', () => {
 			setTimeout(res, 100);
 		});
 		const fakeOutput = mockFullRunData();
-		activeExecutions.remove(executionId, fakeOutput);
+		activeExecutions.finishExecution(executionId, fakeOutput);
 
 		await expect(postExecutePromise).resolves.toEqual(fakeOutput);
 	});
@@ -126,7 +126,7 @@ describe('ActiveExecutions', () => {
 		const cancellablePromise = mockCancelablePromise();
 		cancellablePromise.cancel = cancelExecution;
 		activeExecutions.attachWorkflowExecution(executionId, cancellablePromise);
-		void activeExecutions.stopExecution(executionId);
+		activeExecutions.stopExecution(executionId);
 
 		expect(cancelExecution).toHaveBeenCalledTimes(1);
 	});

packages/cli/src/active-executions.ts

@@ -95,13 +95,23 @@ export class ActiveExecutions {
 			await this.executionRepository.updateExistingExecution(executionId, execution);
 		}
 
+		const postExecutePromise = createDeferredPromise<IRun | undefined>();
+
 		this.activeExecutions[executionId] = {
 			executionData,
 			startedAt: new Date(),
-			postExecutePromises: [],
+			postExecutePromise,
 			status: executionStatus,
 		};
 
+		// Automatically remove execution once the postExecutePromise settles
+		void postExecutePromise.promise.finally(() => {
+			this.concurrencyControl.release({ mode: executionData.executionMode });
+			setImmediate(() => {
+				delete this.activeExecutions[executionId];
+			});
+		});
+
 		return executionId;
 	}
 
@@ -125,68 +135,24 @@ export class ActiveExecutions {
 		execution?.responsePromise?.resolve(response);
 	}
 
-	getPostExecutePromiseCount(executionId: string): number {
+	/** Forces an execution to stop */
-		return this.activeExecutions[executionId]?.postExecutePromises.length ?? 0;
-	}
-
-	/**
-	 * Remove an active execution
-	 */
-	remove(executionId: string, fullRunData?: IRun): void {
-		const execution = this.activeExecutions[executionId];
-		if (execution === undefined) {
-			return;
-		}
-
-		// Resolve all the waiting promises
-		for (const promise of execution.postExecutePromises) {
-			promise.resolve(fullRunData);
-		}
-
-		this.postExecuteCleanup(executionId);
-	}
-
-	/**
-	 * Forces an execution to stop
-	 */
 	stopExecution(executionId: string): void {
-		const execution = this.activeExecutions[executionId];
+		const execution = this.getExecution(executionId);
-		if (execution === undefined) {
+		execution.workflowExecution?.cancel();
-			// There is no execution running with that id
+		execution.postExecutePromise.reject(new ExecutionCancelledError(executionId));
-			return;
-		}
-
-		execution.workflowExecution!.cancel();
-
-		// Reject all the waiting promises
-		const reason = new ExecutionCancelledError(executionId);
-		for (const promise of execution.postExecutePromises) {
-			promise.reject(reason);
-		}
-
-		this.postExecuteCleanup(executionId);
 	}
 
-	private postExecuteCleanup(executionId: string) {
+	/** Mark an execution as completed */
-		const execution = this.activeExecutions[executionId];
+	finishExecution(executionId: string, fullRunData?: IRun) {
-		if (execution === undefined) {
+		const execution = this.getExecution(executionId);
-			return;
+		execution.postExecutePromise.resolve(fullRunData);
-		}
-
-		// Remove from the list of active executions
-		delete this.activeExecutions[executionId];
-
-		this.concurrencyControl.release({ mode: execution.executionData.executionMode });
 	}
 
 	/**
 	 * Returns a promise which will resolve with the data of the execution with the given id
 	 */
 	async getPostExecutePromise(executionId: string): Promise<IRun | undefined> {
-		// Create the promise which will be resolved when the execution finished
+		return await this.getExecution(executionId).postExecutePromise.promise;
-		const waitPromise = createDeferredPromise<IRun | undefined>();
-		this.getExecution(executionId).postExecutePromises.push(waitPromise);
-		return await waitPromise.promise;
 	}
 
 	/**

packages/cli/src/interfaces.ts

@@ -193,7 +193,7 @@ export interface IExecutionsCurrentSummary {
 export interface IExecutingWorkflowData {
 	executionData: IWorkflowExecutionDataProcess;
 	startedAt: Date;
-	postExecutePromises: Array<IDeferredPromise<IRun | undefined>>;
+	postExecutePromise: IDeferredPromise<IRun | undefined>;
 	responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>;
 	workflowExecution?: PCancelable<IRun>;
 	status: ExecutionStatus;

packages/cli/src/workflow-execute-additional-data.ts

@@ -880,7 +880,7 @@ async function executeWorkflow(
 		}
 
 		// remove execution from active executions
-		activeExecutions.remove(executionId, fullRunData);
+		activeExecutions.finishExecution(executionId, fullRunData);
 
 		await executionRepository.updateExistingExecution(executionId, fullExecutionData);
 		throw objectToError(
@@ -906,11 +906,11 @@ async function executeWorkflow(
 	if (data.finished === true || data.status === 'waiting') {
 		// Workflow did finish successfully
 
-		activeExecutions.remove(executionId, data);
+		activeExecutions.finishExecution(executionId, data);
 		const returnData = WorkflowHelpers.getDataLastExecutedNodeData(data);
 		return returnData!.data!.main;
 	}
-	activeExecutions.remove(executionId, data);
+	activeExecutions.finishExecution(executionId, data);
 
 	// Workflow did fail
 	const { error } = data.data.resultData;

packages/cli/src/workflow-runner.ts

@@ -26,7 +26,6 @@ import { ActiveExecutions } from '@/active-executions';
 import config from '@/config';
 import { ExecutionRepository } from '@/databases/repositories/execution.repository';
 import { ExternalHooks } from '@/external-hooks';
-import type { IExecutionResponse } from '@/interfaces';
 import { Logger } from '@/logger';
 import { NodeTypes } from '@/node-types';
 import type { ScalingService } from '@/scaling/scaling.service';
@@ -101,7 +100,7 @@ export class WorkflowRunner {
 
 		// Remove from active execution with empty data. That will
 		// set the execution to failed.
-		this.activeExecutions.remove(executionId, fullRunData);
+		this.activeExecutions.finishExecution(executionId, fullRunData);
 
 		if (hooks) {
 			await hooks.executeHookFunctions('workflowExecuteAfter', [fullRunData]);
@@ -129,7 +128,7 @@ export class WorkflowRunner {
 			await workflowHooks.executeHookFunctions('workflowExecuteBefore', []);
 			await workflowHooks.executeHookFunctions('workflowExecuteAfter', [runData]);
 			responsePromise?.reject(error);
-			this.activeExecutions.remove(executionId);
+			this.activeExecutions.finishExecution(executionId);
 			return executionId;
 		}
 
@@ -321,7 +320,7 @@ export class WorkflowRunner {
 						fullRunData.finished = false;
 					}
 					fullRunData.status = this.activeExecutions.getStatus(executionId);
-					this.activeExecutions.remove(executionId, fullRunData);
+					this.activeExecutions.finishExecution(executionId, fullRunData);
 				})
 				.catch(
 					async (error) =>
@@ -490,45 +489,27 @@ export class WorkflowRunner {
 					reject(error);
 				}
 
-				// optimization: only pull and unflatten execution data from the Db when it is needed
-				const executionHasPostExecutionPromises =
-					this.activeExecutions.getPostExecutePromiseCount(executionId) > 0;
-
-				if (executionHasPostExecutionPromises) {
-					this.logger.debug(
-						`Reading execution data for execution ${executionId} from db for PostExecutionPromise.`,
-					);
-				} else {
-					this.logger.debug(
-						`Skipping execution data for execution ${executionId} since there are no PostExecutionPromise.`,
-					);
-				}
-
 				const fullExecutionData = await this.executionRepository.findSingleExecution(executionId, {
-					includeData: executionHasPostExecutionPromises,
+					includeData: true,
-					unflattenData: executionHasPostExecutionPromises,
+					unflattenData: true,
 				});
 				if (!fullExecutionData) {
 					return reject(new Error(`Could not find execution with id "${executionId}"`));
 				}
 
 				const runData: IRun = {
-					data: {},
 					finished: fullExecutionData.finished,
 					mode: fullExecutionData.mode,
 					startedAt: fullExecutionData.startedAt,
 					stoppedAt: fullExecutionData.stoppedAt,
 					status: fullExecutionData.status,
-				} as IRun;
+					data: fullExecutionData.data,
-
+				};
-				if (executionHasPostExecutionPromises) {
-					runData.data = (fullExecutionData as IExecutionResponse).data;
-				}
 
 				// NOTE: due to the optimization of not loading the execution data from the db when no post execution promises are present,
 				// the execution data in runData.data MAY not be available here.
 				// This means that any function expecting with runData has to check if the runData.data defined from this point
-				this.activeExecutions.remove(executionId, runData);
+				this.activeExecutions.finishExecution(executionId, runData);
 
 				// Normally also static data should be supplied here but as it only used for sending
 				// data to editor-UI is not needed.