fix(core): Aborting manual trigger tests should call closeFunction

packages/workflow/src/Workflow.ts

@@ -1424,13 +1424,6 @@ export class Workflow {
 					return { data: null };
 				}
 
-				if (triggerResponse.manualTriggerFunction !== undefined) {
-					// If a manual trigger function is defined call it and wait till it did run
-					await triggerResponse.manualTriggerFunction();
-				}
-
-				const response = await triggerResponse.manualTriggerResponse!;
-
 				let closeFunction;
 				if (triggerResponse.closeFunction) {
 					// In manual mode we return the trigger closeFunction. That allows it to be called directly
@@ -1439,8 +1432,18 @@ export class Workflow {
 					// If we would not be able to wait for it to close would it cause problems with "own" mode as the
 					// process would be killed directly after it and so the acknowledge would not have been finished yet.
 					closeFunction = triggerResponse.closeFunction;
+
+					// Manual testing of Trigger nodes creates an execution. If the execution is cancelled, `closeFunction` should be called to cleanup any open connections/consumers
+					abortSignal?.addEventListener('abort', closeFunction);
 				}
 
+				if (triggerResponse.manualTriggerFunction !== undefined) {
+					// If a manual trigger function is defined call it and wait till it did run
+					await triggerResponse.manualTriggerFunction();
+				}
+
+				const response = await triggerResponse.manualTriggerResponse!;
+
 				if (response.length === 0) {
 					return { data: null, closeFunction };
 				}

packages/workflow/test/Workflow.test.ts

@@ -4,13 +4,18 @@ import type {
 	IBinaryKeyData,
 	IConnections,
 	IDataObject,
+	IExecuteData,
 	INode,
+	INodeExecuteFunctions,
 	INodeExecutionData,
 	INodeParameters,
 	INodeType,
 	INodeTypeDescription,
 	INodeTypes,
 	IRunExecutionData,
+	ITriggerFunctions,
+	ITriggerResponse,
+	IWorkflowExecuteAdditionalData,
 	NodeParameterValueType,
 } from '@/Interfaces';
 import { Workflow, type WorkflowParameters } from '@/Workflow';
@@ -2015,4 +2020,67 @@ describe('Workflow', () => {
 			]);
 		});
 	});
+
+	describe('runNode', () => {
+		const nodeTypes = mock<INodeTypes>();
+		const triggerNode = mock<INode>();
+		const triggerResponse = mock<ITriggerResponse>({
+			closeFunction: jest.fn(),
+			// This node should never trigger, or return
+			manualTriggerFunction: async () => await new Promise(() => {}),
+		});
+		const triggerNodeType = mock<INodeType>({
+			description: {
+				properties: [],
+			},
+			execute: undefined,
+			poll: undefined,
+			webhook: undefined,
+			async trigger(this: ITriggerFunctions) {
+				return triggerResponse;
+			},
+		});
+
+		nodeTypes.getByNameAndVersion.mockReturnValue(triggerNodeType);
+
+		const workflow = new Workflow({
+			nodeTypes,
+			nodes: [triggerNode],
+			connections: {},
+			active: false,
+		});
+
+		const executionData = mock<IExecuteData>();
+		const runExecutionData = mock<IRunExecutionData>();
+		const additionalData = mock<IWorkflowExecuteAdditionalData>();
+		const nodeExecuteFunctions = mock<INodeExecuteFunctions>();
+		const triggerFunctions = mock<ITriggerFunctions>();
+		nodeExecuteFunctions.getExecuteTriggerFunctions.mockReturnValue(triggerFunctions);
+		const abortController = new AbortController();
+
+		test('should call closeFunction when manual trigger is aborted', async () => {
+			const runPromise = workflow.runNode(
+				executionData,
+				runExecutionData,
+				0,
+				additionalData,
+				nodeExecuteFunctions,
+				'manual',
+				abortController.signal,
+			);
+			// Yield back to the event-loop to let async parts of `runNode` execute
+			await new Promise((resolve) => setImmediate(resolve));
+
+			let isSettled = false;
+			void runPromise.then(() => {
+				isSettled = true;
+			});
+			expect(isSettled).toBe(false);
+			expect(abortController.signal.aborted).toBe(false);
+			expect(triggerResponse.closeFunction).not.toHaveBeenCalled();
+
+			abortController.abort();
+			expect(triggerResponse.closeFunction).toHaveBeenCalled();
+		});
+	});
 });