fix(core): Register workflows as active only after all of the triggers and pollers setup successfully (#12244)

packages/core/src/ActiveWorkflows.ts

@@ -71,16 +71,13 @@ export class ActiveWorkflows {
 		getTriggerFunctions: IGetExecuteTriggerFunctions,
 		getPollFunctions: IGetExecutePollFunctions,
 	) {
-		this.activeWorkflows[workflowId] = {};
 		const triggerNodes = workflow.getTriggerNodes();
 
-		let triggerResponse: ITriggerResponse | undefined;
-
-		this.activeWorkflows[workflowId].triggerResponses = [];
+		const triggerResponses: ITriggerResponse[] = [];
 
 		for (const triggerNode of triggerNodes) {
 			try {
-				triggerResponse = await this.triggersAndPollers.runTrigger(
+				const triggerResponse = await this.triggersAndPollers.runTrigger(
 					workflow,
 					triggerNode,
 					getTriggerFunctions,
@@ -89,10 +86,7 @@ export class ActiveWorkflows {
 					activation,
 				);
 				if (triggerResponse !== undefined) {
-					// If a response was given save it
-
-					// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
-					this.activeWorkflows[workflowId].triggerResponses!.push(triggerResponse);
+					triggerResponses.push(triggerResponse);
 				}
 			} catch (e) {
 				const error = e instanceof Error ? e : new Error(`${e}`);
@@ -104,6 +98,8 @@ export class ActiveWorkflows {
 			}
 		}
 
+		this.activeWorkflows[workflowId] = { triggerResponses };
+
 		const pollingNodes = workflow.getPollNodes();
 
 		if (pollingNodes.length === 0) return;
@@ -119,6 +115,11 @@ export class ActiveWorkflows {
 					activation,
 				);
 			} catch (e) {
+				// Do not mark this workflow as active if there are no triggerResponses, and any polling activation failed
+				if (triggerResponses.length === 0) {
+					delete this.activeWorkflows[workflowId];
+				}
+
 				const error = e instanceof Error ? e : new Error(`${e}`);
 
 				throw new WorkflowActivationError(
@@ -132,7 +133,7 @@ export class ActiveWorkflows {
 	/**
 	 * Activates polling for the given node
 	 */
-	async activatePolling(
+	private async activatePolling(
 		node: INode,
 		workflow: Workflow,
 		additionalData: IWorkflowExecuteAdditionalData,

packages/core/src/__tests__/ActiveWorkflows.test.ts

@@ -0,0 +1,290 @@
+import { mock } from 'jest-mock-extended';
+import type {
+	IGetExecuteTriggerFunctions,
+	INode,
+	ITriggerResponse,
+	IWorkflowExecuteAdditionalData,
+	Workflow,
+	WorkflowActivateMode,
+	WorkflowExecuteMode,
+	TriggerTime,
+	CronExpression,
+} from 'n8n-workflow';
+import { LoggerProxy, TriggerCloseError, WorkflowActivationError } from 'n8n-workflow';
+
+import { ActiveWorkflows } from '@/ActiveWorkflows';
+import type { ErrorReporter } from '@/error-reporter';
+import type { PollContext } from '@/node-execution-context';
+import type { ScheduledTaskManager } from '@/ScheduledTaskManager';
+import type { TriggersAndPollers } from '@/TriggersAndPollers';
+
+describe('ActiveWorkflows', () => {
+	const workflowId = 'test-workflow-id';
+	const workflow = mock<Workflow>();
+	const additionalData = mock<IWorkflowExecuteAdditionalData>();
+	const mode: WorkflowExecuteMode = 'trigger';
+	const activation: WorkflowActivateMode = 'init';
+
+	const getTriggerFunctions = jest.fn() as IGetExecuteTriggerFunctions;
+	const triggerResponse = mock<ITriggerResponse>();
+
+	const pollFunctions = mock<PollContext>();
+	const getPollFunctions = jest.fn<PollContext, unknown[]>();
+
+	LoggerProxy.init(mock());
+	const scheduledTaskManager = mock<ScheduledTaskManager>();
+	const triggersAndPollers = mock<TriggersAndPollers>();
+	const errorReporter = mock<ErrorReporter>();
+	const triggerNode = mock<INode>();
+	const pollNode = mock<INode>();
+
+	let activeWorkflows: ActiveWorkflows;
+
+	beforeEach(() => {
+		jest.clearAllMocks();
+		activeWorkflows = new ActiveWorkflows(scheduledTaskManager, triggersAndPollers, errorReporter);
+	});
+
+	type PollTimes = { item: TriggerTime[] };
+	type TestOptions = {
+		triggerNodes?: INode[];
+		pollNodes?: INode[];
+		triggerError?: Error;
+		pollError?: Error;
+		pollTimes?: PollTimes;
+	};
+
+	const addWorkflow = async ({
+		triggerNodes = [],
+		pollNodes = [],
+		triggerError,
+		pollError,
+		pollTimes = { item: [{ mode: 'everyMinute' }] },
+	}: TestOptions) => {
+		workflow.getTriggerNodes.mockReturnValue(triggerNodes);
+		workflow.getPollNodes.mockReturnValue(pollNodes);
+		pollFunctions.getNodeParameter.calledWith('pollTimes').mockReturnValue(pollTimes);
+
+		if (triggerError) {
+			triggersAndPollers.runTrigger.mockRejectedValueOnce(triggerError);
+		} else {
+			triggersAndPollers.runTrigger.mockResolvedValue(triggerResponse);
+		}
+
+		if (pollError) {
+			triggersAndPollers.runPoll.mockRejectedValueOnce(pollError);
+		} else {
+			getPollFunctions.mockReturnValue(pollFunctions);
+		}
+
+		return await activeWorkflows.add(
+			workflowId,
+			workflow,
+			additionalData,
+			mode,
+			activation,
+			getTriggerFunctions,
+			getPollFunctions,
+		);
+	};
+
+	describe('add()', () => {
+		describe('should activate workflow', () => {
+			it('with trigger nodes', async () => {
+				await addWorkflow({ triggerNodes: [triggerNode] });
+
+				expect(activeWorkflows.isActive(workflowId)).toBe(true);
+				expect(workflow.getTriggerNodes).toHaveBeenCalled();
+				expect(triggersAndPollers.runTrigger).toHaveBeenCalledWith(
+					workflow,
+					triggerNode,
+					getTriggerFunctions,
+					additionalData,
+					mode,
+					activation,
+				);
+			});
+
+			it('with polling nodes', async () => {
+				await addWorkflow({ pollNodes: [pollNode] });
+
+				expect(activeWorkflows.isActive(workflowId)).toBe(true);
+				expect(workflow.getPollNodes).toHaveBeenCalled();
+				expect(scheduledTaskManager.registerCron).toHaveBeenCalled();
+			});
+
+			it('with both trigger and polling nodes', async () => {
+				await addWorkflow({ triggerNodes: [triggerNode], pollNodes: [pollNode] });
+
+				expect(activeWorkflows.isActive(workflowId)).toBe(true);
+				expect(workflow.getTriggerNodes).toHaveBeenCalled();
+				expect(workflow.getPollNodes).toHaveBeenCalled();
+				expect(triggersAndPollers.runTrigger).toHaveBeenCalledWith(
+					workflow,
+					triggerNode,
+					getTriggerFunctions,
+					additionalData,
+					mode,
+					activation,
+				);
+				expect(scheduledTaskManager.registerCron).toHaveBeenCalled();
+				expect(triggersAndPollers.runPoll).toHaveBeenCalledWith(workflow, pollNode, pollFunctions);
+			});
+		});
+
+		describe('should throw error', () => {
+			it('if trigger activation fails', async () => {
+				const error = new Error('Trigger activation failed');
+				await expect(
+					addWorkflow({ triggerNodes: [triggerNode], triggerError: error }),
+				).rejects.toThrow(WorkflowActivationError);
+				expect(activeWorkflows.isActive(workflowId)).toBe(false);
+			});
+
+			it('if polling activation fails', async () => {
+				const error = new Error('Failed to activate polling');
+				await expect(addWorkflow({ pollNodes: [pollNode], pollError: error })).rejects.toThrow(
+					WorkflowActivationError,
+				);
+				expect(activeWorkflows.isActive(workflowId)).toBe(false);
+			});
+
+			it('if the polling interval is too short', async () => {
+				const pollTimes: PollTimes = {
+					item: [
+						{
+							mode: 'custom',
+							cronExpression: '* * * * *' as CronExpression,
+						},
+					],
+				};
+
+				await expect(addWorkflow({ pollNodes: [pollNode], pollTimes })).rejects.toThrow(
+					'The polling interval is too short. It has to be at least a minute.',
+				);
+
+				expect(scheduledTaskManager.registerCron).not.toHaveBeenCalled();
+			});
+		});
+
+		describe('should handle polling errors', () => {
+			it('should throw error when poll fails during initial testing', async () => {
+				const error = new Error('Poll function failed');
+
+				await expect(addWorkflow({ pollNodes: [pollNode], pollError: error })).rejects.toThrow(
+					WorkflowActivationError,
+				);
+
+				expect(triggersAndPollers.runPoll).toHaveBeenCalledWith(workflow, pollNode, pollFunctions);
+				expect(pollFunctions.__emit).not.toHaveBeenCalled();
+				expect(pollFunctions.__emitError).not.toHaveBeenCalled();
+			});
+
+			it('should emit error when poll fails during regular polling', async () => {
+				const error = new Error('Poll function failed');
+				triggersAndPollers.runPoll
+					.mockResolvedValueOnce(null) // Succeed on first call (testing)
+					.mockRejectedValueOnce(error); // Fail on second call (regular polling)
+
+				await addWorkflow({ pollNodes: [pollNode] });
+
+				// Get the executeTrigger function that was registered
+				const registerCronCall = scheduledTaskManager.registerCron.mock.calls[0];
+				const executeTrigger = registerCronCall[2] as () => Promise<void>;
+
+				// Execute the trigger function to simulate a regular poll
+				await executeTrigger();
+
+				expect(triggersAndPollers.runPoll).toHaveBeenCalledTimes(2);
+				expect(pollFunctions.__emit).not.toHaveBeenCalled();
+				expect(pollFunctions.__emitError).toHaveBeenCalledWith(error);
+			});
+		});
+	});
+
+	describe('remove()', () => {
+		const setupForRemoval = async () => {
+			await addWorkflow({ triggerNodes: [triggerNode] });
+			return await activeWorkflows.remove(workflowId);
+		};
+
+		it('should remove an active workflow', async () => {
+			const result = await setupForRemoval();
+
+			expect(result).toBe(true);
+			expect(activeWorkflows.isActive(workflowId)).toBe(false);
+			expect(scheduledTaskManager.deregisterCrons).toHaveBeenCalledWith(workflowId);
+			expect(triggerResponse.closeFunction).toHaveBeenCalled();
+		});
+
+		it('should return false when removing non-existent workflow', async () => {
+			const result = await activeWorkflows.remove('non-existent');
+
+			expect(result).toBe(false);
+			expect(scheduledTaskManager.deregisterCrons).not.toHaveBeenCalled();
+		});
+
+		it('should handle TriggerCloseError when closing trigger', async () => {
+			const triggerCloseError = new TriggerCloseError(triggerNode, { level: 'warning' });
+			(triggerResponse.closeFunction as jest.Mock).mockRejectedValueOnce(triggerCloseError);
+
+			const result = await setupForRemoval();
+
+			expect(result).toBe(true);
+			expect(activeWorkflows.isActive(workflowId)).toBe(false);
+			expect(triggerResponse.closeFunction).toHaveBeenCalled();
+			expect(errorReporter.error).toHaveBeenCalledWith(triggerCloseError, {
+				extra: { workflowId },
+			});
+		});
+
+		it('should throw WorkflowDeactivationError when closeFunction throws regular error', async () => {
+			const error = new Error('Close function failed');
+			(triggerResponse.closeFunction as jest.Mock).mockRejectedValueOnce(error);
+
+			await addWorkflow({ triggerNodes: [triggerNode] });
+
+			await expect(activeWorkflows.remove(workflowId)).rejects.toThrow(
+				`Failed to deactivate trigger of workflow ID "${workflowId}": "Close function failed"`,
+			);
+
+			expect(triggerResponse.closeFunction).toHaveBeenCalled();
+			expect(errorReporter.error).not.toHaveBeenCalled();
+		});
+	});
+
+	describe('get() and isActive()', () => {
+		it('should return workflow data for active workflow', async () => {
+			await addWorkflow({ triggerNodes: [triggerNode] });
+
+			expect(activeWorkflows.isActive(workflowId)).toBe(true);
+			expect(activeWorkflows.get(workflowId)).toBeDefined();
+		});
+
+		it('should return undefined for non-active workflow', () => {
+			expect(activeWorkflows.isActive('non-existent')).toBe(false);
+			expect(activeWorkflows.get('non-existent')).toBeUndefined();
+		});
+	});
+
+	describe('allActiveWorkflows()', () => {
+		it('should return all active workflow IDs', async () => {
+			await addWorkflow({ triggerNodes: [triggerNode] });
+
+			const activeIds = activeWorkflows.allActiveWorkflows();
+
+			expect(activeIds).toEqual([workflowId]);
+		});
+	});
+
+	describe('removeAllTriggerAndPollerBasedWorkflows()', () => {
+		it('should remove all active workflows', async () => {
+			await addWorkflow({ triggerNodes: [triggerNode] });
+
+			await activeWorkflows.removeAllTriggerAndPollerBasedWorkflows();
+
+			expect(activeWorkflows.allActiveWorkflows()).toEqual([]);
+			expect(scheduledTaskManager.deregisterCrons).toHaveBeenCalledWith(workflowId);
+		});
+	});
+});

packages/core/test/TriggersAndPollers.test.ts

@@ -9,6 +9,8 @@ import type {
 	INodeType,
 	INodeTypes,
 	ITriggerFunctions,
+	WorkflowHooks,
+	IRun,
 } from 'n8n-workflow';
 
 import { TriggersAndPollers } from '@/TriggersAndPollers';
@@ -21,11 +23,13 @@ describe('TriggersAndPollers', () => {
 	});
 	const nodeTypes = mock<INodeTypes>();
 	const workflow = mock<Workflow>({ nodeTypes });
+	const hookFunctions = mock<WorkflowHooks['hookFunctions']>({
+		sendResponse: [],
+		workflowExecuteAfter: [],
+	});
 	const additionalData = mock<IWorkflowExecuteAdditionalData>({
 		hooks: {
-			hookFunctions: {
-				sendResponse: [],
-			},
+			hookFunctions,
 		},
 	});
 	const triggersAndPollers = new TriggersAndPollers();
@@ -39,87 +43,80 @@ describe('TriggersAndPollers', () => {
 		const triggerFunctions = mock<ITriggerFunctions>();
 		const getTriggerFunctions = jest.fn().mockReturnValue(triggerFunctions);
 		const triggerFn = jest.fn();
+		const mockEmitData: INodeExecutionData[][] = [[{ json: { data: 'test' } }]];
+
+		const runTriggerHelper = async (mode: 'manual' | 'trigger' = 'trigger') =>
+			await triggersAndPollers.runTrigger(
+				workflow,
+				node,
+				getTriggerFunctions,
+				additionalData,
+				mode,
+				'init',
+			);
 
 		it('should throw error if node type does not have trigger function', async () => {
-			await expect(
-				triggersAndPollers.runTrigger(
-					workflow,
-					node,
-					getTriggerFunctions,
-					additionalData,
-					'trigger',
-					'init',
-				),
-			).rejects.toThrow(ApplicationError);
+			await expect(runTriggerHelper()).rejects.toThrow(ApplicationError);
 		});
 
 		it('should call trigger function in regular mode', async () => {
 			nodeType.trigger = triggerFn;
 			triggerFn.mockResolvedValue({ test: true });
 
-			const result = await triggersAndPollers.runTrigger(
-				workflow,
-				node,
-				getTriggerFunctions,
-				additionalData,
-				'trigger',
-				'init',
-			);
+			const result = await runTriggerHelper();
 
 			expect(triggerFn).toHaveBeenCalled();
 			expect(result).toEqual({ test: true });
 		});
 
-		it('should handle manual mode with promise resolution', async () => {
-			const mockEmitData: INodeExecutionData[][] = [[{ json: { data: 'test' } }]];
-			const mockTriggerResponse = { workflowId: '123' };
+		describe('manual mode', () => {
+			const getMockTriggerFunctions = () => getTriggerFunctions.mock.results[0]?.value;
 
-			nodeType.trigger = triggerFn;
-			triggerFn.mockResolvedValue(mockTriggerResponse);
+			beforeEach(() => {
+				nodeType.trigger = triggerFn;
+				triggerFn.mockResolvedValue({ workflowId: '123' });
+			});
 
-			const result = await triggersAndPollers.runTrigger(
-				workflow,
-				node,
-				getTriggerFunctions,
-				additionalData,
-				'manual',
-				'init',
-			);
+			it('should handle promise resolution', async () => {
+				const result = await runTriggerHelper('manual');
 
-			expect(result).toBeDefined();
-			expect(result?.manualTriggerResponse).toBeInstanceOf(Promise);
+				expect(result?.manualTriggerResponse).toBeInstanceOf(Promise);
+				getMockTriggerFunctions()?.emit?.(mockEmitData);
+			});
 
-			// Simulate emit
-			const mockTriggerFunctions = getTriggerFunctions.mock.results[0]?.value;
-			if (mockTriggerFunctions?.emit) {
-				mockTriggerFunctions.emit(mockEmitData);
-			}
-		});
+			it('should handle error emission', async () => {
+				const testError = new Error('Test error');
+				const result = await runTriggerHelper('manual');
 
-		it('should handle error emission in manual mode', async () => {
-			const testError = new Error('Test error');
+				getMockTriggerFunctions()?.emitError?.(testError);
+				await expect(result?.manualTriggerResponse).rejects.toThrow(testError);
+			});
 
-			nodeType.trigger = triggerFn;
-			triggerFn.mockResolvedValue({});
+			it('should handle response promise', async () => {
+				const responsePromise = { resolve: jest.fn(), reject: jest.fn() };
+				await runTriggerHelper('manual');
 
-			const result = await triggersAndPollers.runTrigger(
-				workflow,
-				node,
-				getTriggerFunctions,
-				additionalData,
-				'manual',
-				'init',
-			);
+				getMockTriggerFunctions()?.emit?.(mockEmitData, responsePromise);
 
-			expect(result?.manualTriggerResponse).toBeInstanceOf(Promise);
+				expect(hookFunctions.sendResponse?.length).toBe(1);
+				await hookFunctions.sendResponse![0]?.({ testResponse: true });
+				expect(responsePromise.resolve).toHaveBeenCalledWith({ testResponse: true });
+			});
 
-			// Simulate error
-			const mockTriggerFunctions = getTriggerFunctions.mock.results[0]?.value;
-			if (mockTriggerFunctions?.emitError) {
-				mockTriggerFunctions.emitError(testError);
-			}
+			it('should handle both response and done promises', async () => {
+				const responsePromise = { resolve: jest.fn(), reject: jest.fn() };
+				const donePromise = { resolve: jest.fn(), reject: jest.fn() };
+				const mockRunData = mock<IRun>({ data: { resultData: { runData: {} } } });
 
-			await expect(result?.manualTriggerResponse).rejects.toThrow(testError);
+				await runTriggerHelper('manual');
+				getMockTriggerFunctions()?.emit?.(mockEmitData, responsePromise, donePromise);
+
+				await hookFunctions.sendResponse![0]?.({ testResponse: true });
+				expect(responsePromise.resolve).toHaveBeenCalledWith({ testResponse: true });
+
+				await hookFunctions.workflowExecuteAfter?.[0]?.(mockRunData, {});
+				expect(donePromise.resolve).toHaveBeenCalledWith(mockRunData);
+			});
 		});
 	});
 
@@ -127,10 +124,11 @@ describe('TriggersAndPollers', () => {
 		const pollFunctions = mock<IPollFunctions>();
 		const pollFn = jest.fn();
 
+		const runPollHelper = async () =>
+			await triggersAndPollers.runPoll(workflow, node, pollFunctions);
+
 		it('should throw error if node type does not have poll function', async () => {
-			await expect(triggersAndPollers.runPoll(workflow, node, pollFunctions)).rejects.toThrow(
-				ApplicationError,
-			);
+			await expect(runPollHelper()).rejects.toThrow(ApplicationError);
 		});
 
 		it('should call poll function and return result', async () => {
@@ -138,7 +136,7 @@ describe('TriggersAndPollers', () => {
 			nodeType.poll = pollFn;
 			pollFn.mockResolvedValue(mockPollResult);
 
-			const result = await triggersAndPollers.runPoll(workflow, node, pollFunctions);
+			const result = await runPollHelper();
 
 			expect(pollFn).toHaveBeenCalled();
 			expect(result).toBe(mockPollResult);
@@ -148,10 +146,18 @@ describe('TriggersAndPollers', () => {
 			nodeType.poll = pollFn;
 			pollFn.mockResolvedValue(null);
 
-			const result = await triggersAndPollers.runPoll(workflow, node, pollFunctions);
+			const result = await runPollHelper();
 
 			expect(pollFn).toHaveBeenCalled();
 			expect(result).toBeNull();
 		});
+
+		it('should propagate errors from poll function', async () => {
+			nodeType.poll = pollFn;
+			pollFn.mockRejectedValue(new Error('Poll function failed'));
+
+			await expect(runPollHelper()).rejects.toThrow('Poll function failed');
+			expect(pollFn).toHaveBeenCalled();
+		});
 	});
 });