fix(core): Handle errors from triggered and polled workflow executions (#35689)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

packages/@n8n/backend-test-utils/src/db/workflows.ts

@@ -191,7 +191,13 @@ export async function createWorkflowWithTrigger(
 				},
 				{
 					id: 'uuid-2',
-					parameters: { triggerTimes: { item: [{ mode: 'everyMinute' }] } },
+					// Deliberately a schedule that cannot fire during a test run. Suites
+					// that don't mock `ActiveWorkflowManager` register this with the real
+					// `ScheduledTaskManager`, and an `everyMinute` tick starts a real
+					// execution that outlives the test that activated the workflow.
+					parameters: {
+						triggerTimes: { item: [{ mode: 'everyMonth', hour: 0, minute: 0, dayOfMonth: 1 }] },
+					},
 					name: 'Cron',
 					type: 'n8n-nodes-base.cron',
 					typeVersion: 1,

packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts

@@ -63,6 +63,14 @@ describe('Execution Lifecycle Hooks', () => {
 	const redactionProxy = mockInstance(ExecutionRedactionServiceProxy);
 	const workflowHookContext = mockInstance(WorkflowHookContextService);
 
+	/**
+	 * The error-workflow dispatch is deliberately fire-and-forget: the hook does
+	 * not await it, and it resolves `WorkflowExecutionService` through a lazy
+	 * import. Drain the microtask queue (this suite fakes timers, so `nextTick`
+	 * rather than `setImmediate`) before asserting on it.
+	 */
+	const flushErrorWorkflowDispatch = async () => await new Promise(process.nextTick);
+
 	const nodeName = 'Test Node';
 	const nodeType = 'n8n-nodes-base.testNode';
 	const nodeId = 'test-node-id';
@@ -1121,6 +1129,7 @@ describe('Execution Lifecycle Hooks', () => {
 
 					await lifecycleHooks.runHook('workflowExecuteAfter', [failedRun, {}]);
 
+					await flushErrorWorkflowDispatch();
 					expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalledWith(
 						errorWorkflow,
 						{
@@ -1178,6 +1187,7 @@ describe('Execution Lifecycle Hooks', () => {
 
 					await lifecycleHooks.runHook('workflowExecuteAfter', [failedRun, {}]);
 
+					await flushErrorWorkflowDispatch();
 					expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalledWith(
 						errorWorkflow,
 						expect.objectContaining({ workflow: { id: workflowId, name: workflowData.name } }),
@@ -1542,6 +1552,7 @@ describe('Execution Lifecycle Hooks', () => {
 
 				await lifecycleHooks.runHook('workflowExecuteAfter', [failedRun, {}]);
 
+				await flushErrorWorkflowDispatch();
 				expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalledWith(
 					errorWorkflow,
 					{
@@ -1596,6 +1607,7 @@ describe('Execution Lifecycle Hooks', () => {
 
 				await lifecycleHooks.runHook('workflowExecuteAfter', [failedRun, {}]);
 
+				await flushErrorWorkflowDispatch();
 				expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalledWith(
 					errorWorkflow,
 					expect.objectContaining({ workflow: { id: workflowId, name: workflowData.name } }),

packages/cli/src/execution-lifecycle/execute-error-workflow.ts

@@ -7,8 +7,26 @@ import type { IRun, IWorkflowBase, WorkflowExecuteMode } from 'n8n-workflow';
 import type { IWorkflowErrorData } from '@/interfaces';
 import { OwnershipService } from '@/services/ownership.service';
 import { UrlService } from '@/services/url.service';
-// eslint-disable-next-line import-x/no-cycle
-import { WorkflowExecutionService } from '@/workflows/workflow-execution.service';
+
+/**
+ * Resolved lazily. A static import would close the cycle
+ * `workflow-runner -> execution-lifecycle-hooks -> execute-error-workflow ->
+ * workflow-execution.service -> workflow-runner`. Enter that graph at the wrong
+ * point (`workflow-runner`, `wait-tracker`, `webhook-helpers` or
+ * `test-webhooks`) and `emitDecoratorMetadata` captures an unresolved paramtype
+ * for `WorkflowExecutionService`'s `workflowRunner`, which DI silently injects
+ * as `undefined` — the next triggered execution then dies with "Cannot read
+ * properties of undefined (reading 'run')".
+ */
+const getWorkflowExecutionService = async () => {
+	// `no-cycle` counts a dynamic import as an edge. It isn't one that matters
+	// here: the import is evaluated at call time, long after every module in the
+	// chain is initialised, so it creates no evaluation-order dependency. That is
+	// the whole reason this is a dynamic import.
+	// eslint-disable-next-line import-x/no-cycle
+	const { WorkflowExecutionService } = await import('@/workflows/workflow-execution.service.js');
+	return Container.get(WorkflowExecutionService);
+};
 
 /**
  * Checks if there was an error and if errorWorkflow or a trigger is defined. If so it collects
@@ -93,8 +111,9 @@ export function executeErrorWorkflow(
 
 			Container.get(OwnershipService)
 				.getWorkflowProjectCached(workflowId)
-				.then((project) => {
-					void Container.get(WorkflowExecutionService).executeErrorWorkflow(
+				.then(async (project) => {
+					const workflowExecutionService = await getWorkflowExecutionService();
+					await workflowExecutionService.executeErrorWorkflow(
 						errorWorkflow,
 						workflowErrorData,
 						project,
@@ -121,12 +140,22 @@ export function executeErrorWorkflow(
 			logger.debug('Start internal error workflow', { executionId, workflowId });
 			void Container.get(OwnershipService)
 				.getWorkflowProjectCached(workflowId)
-				.then((project) => {
-					void Container.get(WorkflowExecutionService).executeErrorWorkflow(
+				.then(async (project) => {
+					const workflowExecutionService = await getWorkflowExecutionService();
+					await workflowExecutionService.executeErrorWorkflow(
 						workflowId,
 						workflowErrorData,
 						project,
 					);
+				})
+				.catch((error: Error) => {
+					Container.get(ErrorReporter).error(error);
+					logger.error(`Could not execute internal ErrorWorkflow for execution ID ${executionId}`, {
+						executionId,
+						workflowId,
+						error,
+						workflowErrorData,
+					});
 				});
 		}
 	}

packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts

@@ -36,6 +36,8 @@ import { getLastExecutedNodeData } from '@/workflow-helpers';
 import { WorkflowHookContextService } from '@/workflow-hook-context.service';
 import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';
 
+// `no-cycle` still reports a cycle here, but only through the dynamic import
+// in `execute-error-workflow`, which creates no evaluation-order edge.
 // eslint-disable-next-line import-x/no-cycle
 import { executeErrorWorkflow } from './execute-error-workflow';
 import { restoreBinaryDataId } from './restore-binary-data-id';

packages/cli/src/workflow-runner.ts

@@ -35,6 +35,8 @@ import PCancelable from 'p-cancelable';
 import { ActiveExecutions } from '@/active-executions';
 import { ExecutionNotFoundError } from '@/errors/execution-not-found-error';
 import { MaxStalledCountError } from '@/errors/max-stalled-count.error';
+// `no-cycle` still reports a cycle here, but only through the dynamic import
+// in `execute-error-workflow`, which creates no evaluation-order edge.
 // eslint-disable-next-line import-x/no-cycle
 import {
 	getLifecycleHooksForRegularMain,

packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts

@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/unbound-method */
 import type { Logger } from '@n8n/backend-common';
 import type { Project, WorkflowEntity } from '@n8n/db';
+import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
 import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
 import { sleep } from '@n8n/utils/sleep';
 import type { ErrorReporter, IGetExecutePollFunctions, StorageConfig } from 'n8n-core';
@@ -61,6 +62,8 @@ describe('TriggerExecutionContextFactory', () => {
 	const nodeTypes = createNodeTypes();
 
 	let factory: TriggerExecutionContextFactory;
+	let logger: Logger;
+	let errorReporter: ErrorReporter;
 
 	beforeEach(() => {
 		vi.clearAllMocks();
@@ -72,12 +75,13 @@ describe('TriggerExecutionContextFactory', () => {
 		);
 
 		scheduleTriggerJobRegistrar.interceptsNode.mockReturnValue(false);
-		const scopedLogger = mock<Logger>();
-		const rootLogger = mock<Logger>({ scoped: vi.fn().mockReturnValue(scopedLogger) });
+		logger = mock<Logger>();
+		const rootLogger = mock<Logger>({ scoped: vi.fn().mockReturnValue(logger) });
+		errorReporter = mock<ErrorReporter>();
 
 		factory = new TriggerExecutionContextFactory(
 			rootLogger,
-			mock<ErrorReporter>(),
+			errorReporter,
 			activeExecutions,
 			eventService,
 			executionService,
@@ -255,6 +259,171 @@ describe('TriggerExecutionContextFactory', () => {
 
 				await expect(donePromise.promise).resolves.toBeUndefined();
 			});
+
+			describe('when the execution fails', () => {
+				// A trigger fires on a timer, so nothing is awaiting `emit`. Any promise
+				// it derives without a terminal handler surfaces as an unhandled
+				// rejection, which fails the whole vitest run even though every
+				// assertion passed. See DEVP-687.
+				const unhandled: unknown[] = [];
+				const captureUnhandled = (reason: unknown) => unhandled.push(reason);
+
+				beforeEach(() => {
+					unhandled.length = 0;
+					process.on('unhandledRejection', captureUnhandled);
+				});
+
+				afterEach(() => {
+					process.off('unhandledRejection', captureUnhandled);
+				});
+
+				/** Let Node reach the checkpoint where it reports unhandled rejections. */
+				const flush = async () => {
+					await sleep(0);
+					await new Promise((resolve) => setImmediate(resolve));
+				};
+
+				const emitWith = (donePromise?: IDeferredPromise<IRun>) => {
+					const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
+					const additionalData = mock<IWorkflowExecuteAdditionalData>();
+					const mode: WorkflowExecuteMode = 'trigger';
+					const activation: WorkflowActivateMode = 'activate';
+					const workflow = mock<Workflow>({ name: 'Test Workflow' });
+					const node = mock<INode>({ name: 'Trigger Node', id: 'node-1' });
+
+					const getTriggerFunctions = factory.getExecuteTriggerFunctions(
+						workflowData,
+						additionalData,
+						mode,
+						activation,
+						async () => workflowData,
+						vi.fn(),
+						scheduleCollectionSession,
+					);
+					const context = getTriggerFunctions(workflow, node, additionalData, mode, activation);
+
+					context.emit([[]], undefined, donePromise);
+				};
+
+				test('logs the error and does not emit workflow-executed', async () => {
+					workflowExecutionService.runWorkflow.mockRejectedValueOnce(new UnexpectedError('boom'));
+
+					emitWith();
+					await flush();
+
+					expect(unhandled).toEqual([]);
+					expect(logger.error).toHaveBeenCalledWith('boom', expect.objectContaining({}));
+					expect(eventService.emit).not.toHaveBeenCalled();
+				});
+
+				test('rejects donePromise instead of leaving it pending', async () => {
+					workflowExecutionService.runWorkflow.mockRejectedValueOnce(new UnexpectedError('boom'));
+					const donePromise = createDeferredPromise<IRun>();
+
+					emitWith(donePromise);
+
+					await expect(donePromise.promise).rejects.toThrow('boom');
+					await flush();
+					expect(unhandled).toEqual([]);
+				});
+
+				test('rejects donePromise when the workflow data cannot be resolved', async () => {
+					const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
+					const donePromise = createDeferredPromise<IRun>();
+
+					const getTriggerFunctions = factory.getExecuteTriggerFunctions(
+						workflowData,
+						mock<IWorkflowExecuteAdditionalData>(),
+						'trigger',
+						'activate',
+						async () => {
+							throw new UnexpectedError('Published version not found for workflow');
+						},
+						vi.fn(),
+						scheduleCollectionSession,
+					);
+					const context = getTriggerFunctions(
+						mock<Workflow>({ name: 'Test Workflow' }),
+						mock<INode>({ name: 'Trigger Node' }),
+						mock<IWorkflowExecuteAdditionalData>(),
+						'trigger',
+						'activate',
+					);
+
+					context.emit([[]], undefined, donePromise);
+
+					await expect(donePromise.promise).rejects.toThrow('Published version not found');
+					await flush();
+					expect(unhandled).toEqual([]);
+					expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
+				});
+
+				test('rejects donePromise with an Error when the failure is not one', async () => {
+					workflowExecutionService.runWorkflow.mockRejectedValueOnce('just a string');
+					const donePromise = createDeferredPromise<IRun>();
+
+					emitWith(donePromise);
+
+					await expect(donePromise.promise).rejects.toBeInstanceOf(Error);
+					await flush();
+					expect(unhandled).toEqual([]);
+				});
+
+				test('rejects donePromise when the post-execute promise fails', async () => {
+					activeExecutions.getPostExecutePromise.mockRejectedValueOnce(
+						new UnexpectedError('execution gone'),
+					);
+					const donePromise = createDeferredPromise<IRun>();
+
+					emitWith(donePromise);
+
+					await expect(donePromise.promise).rejects.toThrow('execution gone');
+					await flush();
+					expect(unhandled).toEqual([]);
+				});
+
+				test('does not report an unhandled rejection on the happy path', async () => {
+					activeExecutions.getPostExecutePromise.mockResolvedValue(mock<IRun>());
+					const donePromise = createDeferredPromise<IRun>();
+
+					emitWith(donePromise);
+					await flush();
+
+					expect(unhandled).toEqual([]);
+					expect(eventService.emit).toHaveBeenCalledTimes(1);
+				});
+
+				test('logs when the failed execution cannot be recorded', async () => {
+					vi.spyOn(factory, 'executeErrorWorkflow').mockImplementation(() => {});
+					executionService.createErrorExecution.mockRejectedValueOnce(
+						new UnexpectedError('db down'),
+					);
+
+					const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
+					const getTriggerFunctions = factory.getExecuteTriggerFunctions(
+						workflowData,
+						mock<IWorkflowExecuteAdditionalData>(),
+						'trigger',
+						'activate',
+						async () => workflowData,
+						vi.fn(),
+						scheduleCollectionSession,
+					);
+					const context = getTriggerFunctions(
+						mock<Workflow>({ name: 'Test Workflow' }),
+						mock<INode>({ name: 'Trigger Node' }),
+						mock<IWorkflowExecuteAdditionalData>(),
+						'trigger',
+						'activate',
+					);
+
+					context.saveFailedExecution(mock<ExecutionError>());
+					await flush();
+
+					expect(unhandled).toEqual([]);
+					expect(errorReporter.error).toHaveBeenCalled();
+				});
+			});
 		});
 
 		describe('emitError', () => {
@@ -455,6 +624,98 @@ describe('TriggerExecutionContextFactory', () => {
 				expect(activeExecutions.getPostExecutePromise).toHaveBeenCalledWith('exec-123');
 				await expect(donePromise.promise).resolves.toBe(runResult);
 			});
+
+			describe('when the execution fails', () => {
+				// See the equivalent block for `emit` — a poll fires on a timer too.
+				const unhandled: unknown[] = [];
+				const captureUnhandled = (reason: unknown) => unhandled.push(reason);
+
+				beforeEach(() => {
+					unhandled.length = 0;
+					process.on('unhandledRejection', captureUnhandled);
+				});
+
+				afterEach(() => {
+					process.off('unhandledRejection', captureUnhandled);
+				});
+
+				const flush = async () => {
+					await sleep(0);
+					await new Promise((resolve) => setImmediate(resolve));
+				};
+
+				const pollEmitWith = (donePromise?: IDeferredPromise<IRun>) => {
+					const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
+					const additionalData = mock<IWorkflowExecuteAdditionalData>();
+
+					const getPollFunctions = factory.getExecutePollFunctions(
+						workflowData,
+						additionalData,
+						'trigger',
+						'activate',
+						async () => workflowData,
+					);
+					const context = getPollFunctions(
+						mock<Workflow>({ id: 'wf-1', name: 'Test Workflow' }),
+						mock<INode>({ name: 'Poll Node' }),
+						additionalData,
+						'trigger',
+						'activate',
+					);
+
+					context.__emit([[]], undefined, donePromise);
+				};
+
+				test('logs the error without a donePromise', async () => {
+					workflowExecutionService.runWorkflow.mockRejectedValueOnce(new UnexpectedError('boom'));
+
+					pollEmitWith();
+					await flush();
+
+					expect(unhandled).toEqual([]);
+					expect(logger.error).toHaveBeenCalledWith('boom', expect.objectContaining({}));
+				});
+
+				test('rejects donePromise instead of leaving it pending', async () => {
+					workflowExecutionService.runWorkflow.mockRejectedValueOnce(new UnexpectedError('boom'));
+					const donePromise = createDeferredPromise<IRun>();
+
+					pollEmitWith(donePromise);
+
+					await expect(donePromise.promise).rejects.toThrow('boom');
+					await flush();
+					expect(unhandled).toEqual([]);
+				});
+
+				test('logs when the failed execution cannot be recorded', async () => {
+					vi.spyOn(factory, 'executeErrorWorkflow').mockImplementation(() => {});
+					executionService.createErrorExecution.mockRejectedValueOnce(
+						new UnexpectedError('db down'),
+					);
+
+					const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
+					const getPollFunctions = factory.getExecutePollFunctions(
+						workflowData,
+						mock<IWorkflowExecuteAdditionalData>(),
+						'trigger',
+						'activate',
+						async () => workflowData,
+					);
+					const context = getPollFunctions(
+						mock<Workflow>({ id: 'wf-1', name: 'Test Workflow' }),
+						mock<INode>({ name: 'Poll Node' }),
+						mock<IWorkflowExecuteAdditionalData>(),
+						'trigger',
+						'activate',
+					);
+
+					context.__emitError(mock<ExecutionError>());
+					await flush();
+
+					expect(unhandled).toEqual([]);
+					expect(errorReporter.error).toHaveBeenCalled();
+				});
+			});
 		});
 
 		describe('__emitError', () => {

packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts

@@ -1,6 +1,7 @@
 import { Logger } from '@n8n/backend-common';
 import type { IWorkflowDb } from '@n8n/db';
 import { Service } from '@n8n/di';
+import { ensureError } from '@n8n/utils/errors/ensure-error';
 import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
 import {
 	ErrorReporter,
@@ -76,6 +77,65 @@ export class TriggerExecutionContextFactory {
 		this.logger = this.logger.scoped(['workflow-activation']);
 	}
 
+	/**
+	 * Bridge an in-flight triggered execution to the node's `donePromise`. It
+	 * always settles: with the finished run, with `undefined` when a duplicate
+	 * scheduled execution was skipped, or rejected if the execution never
+	 * started. Nodes that await it (MQTT, RabbitMQ, AMQP, Kafka) would otherwise
+	 * hang forever on failure.
+	 */
+	private settleDonePromise(
+		executePromise: Promise<string | undefined>,
+		donePromise: IDeferredPromise<IRun | undefined>,
+	) {
+		void executePromise
+			.then(async (executionId) =>
+				executionId === undefined
+					? undefined
+					: await this.activeExecutions.getPostExecutePromise(executionId),
+			)
+			.then(donePromise.resolve, (error: unknown) => donePromise.reject(ensureError(error)));
+	}
+
+	/**
+	 * Terminates the emit promise chains. Without it a failed triggered execution
+	 * surfaces as an unhandled rejection instead of a logged error.
+	 */
+	private logTriggerExecutionFailure(error: unknown, workflowData: IWorkflowBase, node: INode) {
+		const failure = ensureError(error);
+		this.logger.error(failure.message, {
+			error: failure,
+			workflowId: workflowData.id,
+			nodeName: node.name,
+		});
+	}
+
+	/**
+	 * Persist a failed trigger execution, then run the error workflow for it.
+	 */
+	private recordTriggerFailure(
+		error: ExecutionError,
+		node: INode,
+		workflowData: IWorkflowBase,
+		workflow: Workflow,
+		mode: WorkflowExecuteMode,
+	) {
+		void this.executionService
+			.createErrorExecution(error, node, workflowData, workflow, mode)
+			.then(() => {
+				this.executeErrorWorkflow(error, workflowData, mode);
+			})
+			.catch((cause: unknown) => {
+				const failure = ensureError(cause);
+				this.errorReporter.error(failure);
+				this.logger.error('Failed to record failed trigger execution', {
+					error: failure,
+					workflowId: workflowData.id,
+					nodeName: node.name,
+				});
+			});
+	}
+
 	/**
 	 * Return trigger function which gets the global functions from n8n-core
 	 * and overwrites the emit to be able to start it in subprocess
@@ -136,40 +196,29 @@ export class TriggerExecutionContextFactory {
 						throw error;
 					});
 
-				void executePromise.then(async (executionId) => {
-					// `executionId` is undefined when the catch above swallowed a
-					// duplicate scheduled execution; nothing ran, so nothing to emit.
-					if (executionId === undefined) return;
-					const { projectId, projectName } = await getWorkflowProjectDetailsSafe(
-						this.ownershipService,
-						workflowData.id,
-					);
-					this.eventService.emit('workflow-executed', {
-						workflowId: workflowData.id,
-						workflowName: workflowData.name,
-						executionId,
-						projectId,
-						projectName,
-						source: 'trigger',
-					});
-				});
+				// Registered ahead of the telemetry chain so the post-execute promise
+				// is attached before that chain's ownership lookup yields.
+				if (donePromise) this.settleDonePromise(executePromise, donePromise);
 
-				if (donePromise) {
-					void executePromise.then((executionId) => {
-						// Same as above: a duplicate scheduled execution was skipped,
-						// so resolve with undefined and don't wait on a non-existent run.
-						if (executionId === undefined) {
-							donePromise.resolve(undefined);
-							return;
-						}
-						this.activeExecutions
-							.getPostExecutePromise(executionId)
-							.then(donePromise.resolve)
-							.catch(donePromise.reject);
-					});
-				} else {
-					executePromise.catch((error: Error) => this.logger.error(error.message, { error }));
-				}
+				void executePromise
+					.then(async (executionId) => {
+						// `executionId` is undefined when the catch above swallowed a
+						// duplicate scheduled execution; nothing ran, so nothing to emit.
+						if (executionId === undefined) return;
+						const { projectId, projectName } = await getWorkflowProjectDetailsSafe(
+							this.ownershipService,
+							workflowData.id,
+						);
+						this.eventService.emit('workflow-executed', {
+							workflowId: workflowData.id,
+							workflowName: workflowData.name,
+							executionId,
+							projectId,
+							projectName,
+							source: 'trigger',
+						});
+					})
+					.catch((error: unknown) => this.logTriggerExecutionFailure(error, workflowData, node));
 			};
 
 			const emitError = (error: Error): void => {
@@ -185,11 +234,7 @@ export class TriggerExecutionContextFactory {
 						workflowName: workflowData.name,
 					},
 				);
-				void this.executionService
-					.createErrorExecution(error, node, workflowData, workflow, mode)
-					.then(() => {
-						this.executeErrorWorkflow(error, workflowData, mode);
-					});
+				this.recordTriggerFailure(error, node, workflowData, workflow, mode);
 			};
 
 			const schedulingFunctions = this.scheduleTriggerJobRegistrar.interceptsNode(node)
@@ -249,24 +294,15 @@ export class TriggerExecutionContextFactory {
 						),
 				);
 
-				if (donePromise) {
-					void executePromise.then((executionId) => {
-						this.activeExecutions
-							.getPostExecutePromise(executionId)
-							.then(donePromise.resolve)
-							.catch(donePromise.reject);
-					});
-				} else {
-					void executePromise.catch((error: Error) => this.logger.error(error.message, { error }));
-				}
+				if (donePromise) this.settleDonePromise(executePromise, donePromise);
+
+				void executePromise.catch((error: unknown) =>
+					this.logTriggerExecutionFailure(error, workflowData, node),
+				);
 			};
 
 			const __emitError = (error: ExecutionError) => {
-				void this.executionService
-					.createErrorExecution(error, node, workflowData, workflow, mode)
-					.then(() => {
-						this.executeErrorWorkflow(error, workflowData, mode);
-					});
+				this.recordTriggerFailure(error, node, workflowData, workflow, mode);
 			};
 
 			return new PollContext(workflow, node, additionalData, mode, activation, __emit, __emitError);

packages/cli/test/integration/webhooks.api.test.ts

@@ -87,6 +87,7 @@ describe('Webhook API', () => {
 	let user: User;
 	let agent: SuperAgentTest;
 	let workflow: WorkflowEntity | undefined;
+	let activeWorkflowManager: Awaited<ReturnType<typeof initActiveWorkflowManager>> | undefined;
 
 	beforeAll(async () => {
 		await testDb.init();
@@ -100,10 +101,13 @@ describe('Webhook API', () => {
 	beforeEach(async () => {
 		await testDb.truncate(['WorkflowEntity']);
 		workflow = await createActiveWorkflow(workflowData, user);
-		await initActiveWorkflowManager();
+		activeWorkflowManager = await initActiveWorkflowManager();
 	});
 
 	afterEach(async () => {
+		// The manager is re-inited per test, so without this each run leaves its
+		// registrations live for the rest of the worker.
+		await activeWorkflowManager?.removeAll();
 		if (workflow) {
 			await deleteWorkflowAndWebhooks(workflow.id);
 		}

packages/cli/test/integration/workflows/workflows.controller-with-active-workflow-manager.ee.test.ts

@@ -5,7 +5,9 @@ import {
 	createActiveWorkflow,
 } from '@n8n/backend-test-utils';
 import type { User } from '@n8n/db';
+import { Container } from '@n8n/di';
 
+import { ActiveWorkflowManager } from '@/active-workflow-manager';
 import { Telemetry } from '@/telemetry';
 
 import { createUser } from '../shared/db/users';
@@ -35,6 +37,12 @@ beforeEach(async () => {
 	]);
 });
 
+afterEach(async () => {
+	// This suite drives the real ActiveWorkflowManager, so anything it registers
+	// (cron jobs, webhooks) would otherwise stay live for the rest of the worker.
+	await Container.get(ActiveWorkflowManager).removeAll();
+});
+
 describe('PUT /:workflowId/transfer', () => {
 	// This tests does not mock the ActiveWorkflowManager, which helps catching
 	// possible deadlocks when using transactions wrong.

packages/cli/test/unit/no-source-import-cycles.test.ts

@@ -0,0 +1,150 @@
+import { readdirSync, readFileSync, statSync } from 'node:fs';
+import path from 'node:path';
+
+/**
+ * A value-import cycle between two `@Service()` modules is not a style problem:
+ * whichever module the graph is entered at gets an unresolved `design:paramtypes`
+ * entry for the other, and `@n8n/di` silently injects `undefined` for it. The
+ * class then fails at call time with "Cannot read properties of undefined",
+ * hundreds of test files away from the import that caused it.
+ *
+ * `import-x/no-cycle` is only a warning, and it reports per-import rather than
+ * per-cycle, so this test is the gate.
+ */
+
+const SRC = path.resolve(__dirname, '../../src');
+
+const resolveSpecifier = (specifier: string, importer: string): string | null => {
+	let base: string;
+	if (specifier.startsWith('@/')) base = path.join(SRC, specifier.slice(2));
+	else if (specifier.startsWith('.')) base = path.resolve(path.dirname(importer), specifier);
+	else return null;
+
+	// `@/foo.js` is the on-disk `@/foo.ts` (NodeNext-style specifiers).
+	const withoutJs = base.endsWith('.js') ? base.slice(0, -3) : base;
+
+	for (const candidate of [withoutJs + '.ts', path.join(withoutJs, 'index.ts')]) {
+		try {
+			if (statSync(candidate).isFile()) return candidate;
+		} catch {
+			// not this candidate
+		}
+	}
+	return null;
+};
+
+/**
+ * Static value imports and re-exports only. `import type` and dynamic
+ * `await import()` do not create an evaluation-order edge, which is exactly
+ * why they are the escape hatch when a cycle is unavoidable.
+ */
+// Clauses never contain quotes or semicolons, so [^'";]*? spans multi-line
+// braced imports without leaking past a bare side-effect import's statement.
+const importPattern = /(?:^|\n)\s*import\s+(type\s+)?(?:[^'";]*?\bfrom\s*)?['"]([^'"]+)['"]/g;
+// Re-exports (`export ... from`) are evaluation-order edges too — barrels route cycles.
+const exportPattern =
+	/(?:^|\n)\s*export\s+(type\s+)?(?:\*(?:\s+as\s+[\w$]+)?|\{[^}]*\})\s*from\s*['"]([^'"]+)['"]/g;
+
+const valueImportsOf = (file: string): string[] => {
+	const source = readFileSync(file, 'utf8');
+	const edges: string[] = [];
+
+	for (const pattern of [importPattern, exportPattern]) {
+		for (const match of source.matchAll(pattern)) {
+			if (match[1]) continue;
+			const resolved = resolveSpecifier(match[2], file);
+			if (resolved) edges.push(resolved);
+		}
+	}
+	return edges;
+};
+
+/** Tarjan's algorithm, iterative to stay clear of the call-stack limit. */
+const findCycles = (nodes: string[], edgesOf: (node: string) => string[]): string[][] => {
+	const index = new Map<string, number>();
+	const lowlink = new Map<string, number>();
+	const onStack = new Set<string>();
+	const stack: string[] = [];
+	const cycles: string[][] = [];
+	let counter = 0;
+
+	for (const root of nodes) {
+		if (index.has(root)) continue;
+
+		const work: Array<{ node: string; edge: number }> = [{ node: root, edge: 0 }];
+		index.set(root, counter);
+		lowlink.set(root, counter);
+		counter += 1;
+		stack.push(root);
+		onStack.add(root);
+
+		while (work.length > 0) {
+			const frame = work[work.length - 1];
+			const neighbours = edgesOf(frame.node);
+
+			if (frame.edge < neighbours.length) {
+				const next = neighbours[frame.edge];
+				frame.edge += 1;
+
+				if (!index.has(next)) {
+					index.set(next, counter);
+					lowlink.set(next, counter);
+					counter += 1;
+					stack.push(next);
+					onStack.add(next);
+					work.push({ node: next, edge: 0 });
+				} else if (onStack.has(next)) {
+					lowlink.set(frame.node, Math.min(lowlink.get(frame.node)!, index.get(next)!));
+				}
+				continue;
+			}
+
+			work.pop();
+			const parent = work[work.length - 1];
+			if (parent) {
+				lowlink.set(parent.node, Math.min(lowlink.get(parent.node)!, lowlink.get(frame.node)!));
+			}
+
+			if (lowlink.get(frame.node) === index.get(frame.node)) {
+				const component: string[] = [];
+				let member: string;
+				do {
+					member = stack.pop()!;
+					onStack.delete(member);
+					component.push(member);
+				} while (member !== frame.node);
+
+				if (component.length > 1) cycles.push(component);
+			}
+		}
+	}
+
+	return cycles;
+};
+
+test('packages/cli/src has no value-import cycles', () => {
+	const files = readdirSync(SRC, { recursive: true, encoding: 'utf8' })
+		.filter(
+			(file) => file.endsWith('.ts') && !file.includes('__tests__') && !file.endsWith('.test.ts'),
+		)
+		.map((file) => path.join(SRC, file));
+
+	const cache = new Map<string, string[]>();
+	const edgesOf = (file: string) => {
+		let edges = cache.get(file);
+		if (!edges) {
+			edges = valueImportsOf(file);
+			cache.set(file, edges);
+		}
+		return edges;
+	};
+
+	const cycles = findCycles(files, edgesOf)
+		// Only a cycle that reaches a DI service can corrupt an injected
+		// dependency. TypeORM entities, for one, reference each other by design
+		// for bidirectional relations and have no injected constructor params.
+		.filter((cycle) => cycle.some((file) => readFileSync(file, 'utf8').includes('@Service(')))
+		.map((cycle) => cycle.map((file) => path.relative(SRC, file)).sort());
+
+	expect(cycles).toEqual([]);
+});