refactor(core): Give poll trigger execution its own scoped logger (#32986)

packages/@n8n/config/src/configs/logging.config.ts

@@ -47,6 +47,7 @@ export const LOG_SCOPES = [
 	'oauth-jwe',
 	'mcp-registry',
 	'workflow-publication',
+	'poll-trigger',
 	'metrics',
 ] as const;
 

packages/cli/src/__tests__/active-workflow-manager.test.ts

@@ -5,7 +5,12 @@ import type { WorkflowsConfig } from '@n8n/config';
 import type { WorkflowEntity, WorkflowHistory, WorkflowRepository } from '@n8n/db';
 import { mock } from 'jest-mock-extended';
 import type { InstanceSettings } from 'n8n-core';
-import { ActiveWorkflowTriggers, ScheduledTaskManager } from 'n8n-core';
+import {
+	ActiveWorkflowTriggers,
+	PollTriggerExecutor,
+	ScheduledTaskManager,
+	Tracing,
+} from 'n8n-core';
 import type {
 	CronExpression,
 	ExecutionError,
@@ -678,22 +683,19 @@ describe('ActiveWorkflowManager', () => {
 
 				const workflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
 				const additionalData = mock<IWorkflowExecuteAdditionalData>();
+				const logger = mock<Logger>({ scoped: jest.fn().mockReturnValue(mock<Logger>()) });
+				const triggersAndPollers = {
+					runPollFunction: async (wf: Workflow, node: INode, pollFunctions: IPollFunctions) =>
+						await wf.nodeTypes
+							.getByNameAndVersion(node.type, node.typeVersion)
+							.poll!.call(pollFunctions),
+				} as ConstructorParameters<typeof ActiveWorkflowTriggers>[2];
 				const realActiveWorkflowTriggers = new ActiveWorkflowTriggers(
-					mock<Logger>({ scoped: jest.fn().mockReturnValue(mock<Logger>()) }),
+					logger,
 					scheduledTaskManager,
-					{
-						runPollFunction: async (wf: Workflow, node: INode, pollFunctions: IPollFunctions) =>
-							await wf.nodeTypes
-								.getByNameAndVersion(node.type, node.typeVersion)
-								.poll!.call(pollFunctions),
-					} as ConstructorParameters<typeof ActiveWorkflowTriggers>[2],
+					triggersAndPollers,
 					mock(),
-					{
-						startSpan: async (_options: unknown, fn: (span: unknown) => Promise<void>) =>
-							await fn({ setStatus: () => {} }),
-						pickWorkflowAttributes: () => ({}),
-						pickNodeAttributes: () => ({}),
-					} as unknown as ConstructorParameters<typeof ActiveWorkflowTriggers>[4],
+					new PollTriggerExecutor(logger, triggersAndPollers, new Tracing()),
 				);
 
 				await realActiveWorkflowTriggers.addAllTriggers(
@@ -758,7 +760,7 @@ describe('ActiveWorkflowManager', () => {
 				realScheduledTaskManager,
 				mock(),
 				mock(),
-				mock(),
+				mock<PollTriggerExecutor>(),
 			);
 			activeWorkflowManager = new ActiveWorkflowManager(
 				mockLogger(),

packages/core/src/execution-engine/__tests__/active-workflow-triggers.test.ts

@@ -26,6 +26,7 @@ import { Tracing } from '@/observability';
 import { ActiveWorkflowTriggers } from '../active-workflow-triggers';
 import type { IGetExecuteTriggerFunctions } from '../interfaces';
 import type { PollContext } from '../node-execution-context';
+import { PollTriggerExecutor } from '../poll-trigger-executor';
 import { ScheduledTaskManager } from '../scheduled-task-manager';
 import type { TriggersAndPollers } from '../triggers-and-pollers';
 
@@ -79,7 +80,7 @@ describe('ActiveWorkflowTriggers', () => {
 			scheduledTaskManager,
 			triggersAndPollers,
 			errorReporter,
-			tracing,
+			new PollTriggerExecutor(logger, triggersAndPollers, tracing),
 		);
 	});
 
@@ -1202,7 +1203,7 @@ describe('ActiveWorkflowTriggers', () => {
 				realScheduledTaskManager,
 				triggersAndPollers,
 				errorReporter,
-				tracing,
+				new PollTriggerExecutor(realLogger, triggersAndPollers, tracing),
 			);
 		});
 

packages/core/src/execution-engine/__tests__/poll-trigger-executor.test.ts

@@ -0,0 +1,146 @@
+import type { Logger } from '@n8n/backend-common';
+import type { INode, INodeExecutionData, Workflow } from 'n8n-workflow';
+import { LoggerProxy } from 'n8n-workflow';
+import type { Mock } from 'vitest';
+import { mock } from 'vitest-mock-extended';
+
+import { Tracing } from '@/observability';
+
+import type { PollContext } from '../node-execution-context';
+import { PollTriggerExecutor } from '../poll-trigger-executor';
+import type { TriggersAndPollers } from '../triggers-and-pollers';
+
+describe('PollTriggerExecutor', () => {
+	const tracing = new Tracing();
+
+	LoggerProxy.init(mock());
+	const logger = mock<Logger>();
+	logger.scoped.mockReturnValue(logger);
+
+	const triggersAndPollers = mock<TriggersAndPollers>();
+	const node = mock<INode>({ id: 'poll-node', name: 'Poll Node' });
+	const pollFunctions = mock<PollContext>();
+
+	let executor: PollTriggerExecutor;
+	let workflow: Workflow;
+	let acquireIsolate: Mock;
+	let releaseIsolate: Mock;
+
+	beforeEach(() => {
+		vi.clearAllMocks();
+		logger.scoped.mockReturnValue(logger);
+		acquireIsolate = vi.fn().mockResolvedValue(undefined);
+		releaseIsolate = vi.fn().mockResolvedValue(undefined);
+		workflow = mock<Workflow>({ id: 'wf-id', name: 'My Workflow' });
+		// @ts-expect-error -- minimal expression stub for isolate-acquisition tests
+		workflow.expression = { acquireIsolate, releaseIsolate };
+		executor = new PollTriggerExecutor(logger, triggersAndPollers, tracing);
+	});
+
+	it('uses a logger scoped to "poll-trigger"', () => {
+		expect(logger.scoped).toHaveBeenCalledWith('poll-trigger');
+	});
+
+	describe('initial activation poll (testingTrigger=true)', () => {
+		it('emits the poll result without acquiring the isolate', async () => {
+			const result: INodeExecutionData[][] = [[{ json: { ok: true } }]];
+			triggersAndPollers.runPollFunction.mockResolvedValueOnce(result);
+
+			const execute = executor.create(workflow, node, pollFunctions, () => true);
+			await execute(true);
+
+			expect(triggersAndPollers.runPollFunction).toHaveBeenCalledWith(
+				workflow,
+				node,
+				pollFunctions,
+			);
+			expect(pollFunctions.__emit).toHaveBeenCalledWith(result);
+			// The initial poll runs inside the outer isolate window, so it must not acquire its own.
+			expect(acquireIsolate).not.toHaveBeenCalled();
+			expect(releaseIsolate).not.toHaveBeenCalled();
+		});
+
+		it('does not emit when the poll returns null', async () => {
+			triggersAndPollers.runPollFunction.mockResolvedValueOnce(null);
+
+			const execute = executor.create(workflow, node, pollFunctions, () => true);
+			await execute(true);
+
+			expect(pollFunctions.__emit).not.toHaveBeenCalled();
+		});
+
+		it('rethrows the poll error so activation fails', async () => {
+			const error = new Error('poll failed');
+			triggersAndPollers.runPollFunction.mockRejectedValueOnce(error);
+
+			const execute = executor.create(workflow, node, pollFunctions, () => true);
+
+			await expect(execute(true)).rejects.toThrow(error);
+			expect(pollFunctions.__emitError).not.toHaveBeenCalled();
+		});
+	});
+
+	describe('scheduled poll (testingTrigger=false)', () => {
+		it('acquires and releases the isolate and emits the result', async () => {
+			const result: INodeExecutionData[][] = [[{ json: { ok: true } }]];
+			triggersAndPollers.runPollFunction.mockResolvedValueOnce(result);
+
+			const execute = executor.create(workflow, node, pollFunctions, () => true);
+			await execute();
+
+			expect(acquireIsolate).toHaveBeenCalledTimes(1);
+			expect(pollFunctions.__emit).toHaveBeenCalledWith(result);
+			expect(releaseIsolate).toHaveBeenCalledTimes(1);
+		});
+
+		it('emits an error when the poll fails for a current workflow', async () => {
+			const error = new Error('poll failed');
+			triggersAndPollers.runPollFunction.mockRejectedValueOnce(error);
+
+			const execute = executor.create(workflow, node, pollFunctions, () => true);
+			await execute();
+
+			expect(pollFunctions.__emitError).toHaveBeenCalledWith(error);
+			expect(releaseIsolate).toHaveBeenCalledTimes(1);
+		});
+
+		it('skips the poll entirely when superseded before running', async () => {
+			const execute = executor.create(workflow, node, pollFunctions, () => false);
+			await execute();
+
+			expect(triggersAndPollers.runPollFunction).not.toHaveBeenCalled();
+			expect(pollFunctions.__emit).not.toHaveBeenCalled();
+			expect(acquireIsolate).not.toHaveBeenCalled();
+		});
+
+		it('drops an in-flight result when superseded after the poll resolves', async () => {
+			let isCurrent = true;
+			triggersAndPollers.runPollFunction.mockImplementationOnce(async () => {
+				// The workflow is removed/reactivated while the poll is in flight.
+				isCurrent = false;
+				return [[{ json: { stale: true } }]];
+			});
+
+			const execute = executor.create(workflow, node, pollFunctions, () => isCurrent);
+			await execute();
+
+			expect(pollFunctions.__emit).not.toHaveBeenCalled();
+			// The dropped poll still releases the isolate it acquired.
+			expect(releaseIsolate).toHaveBeenCalledTimes(1);
+		});
+
+		it('ignores a poll error when superseded', async () => {
+			let isCurrent = true;
+			triggersAndPollers.runPollFunction.mockImplementationOnce(async () => {
+				isCurrent = false;
+				throw new Error('poll failed');
+			});
+
+			const execute = executor.create(workflow, node, pollFunctions, () => isCurrent);
+			await execute();
+
+			expect(pollFunctions.__emitError).not.toHaveBeenCalled();
+			expect(releaseIsolate).toHaveBeenCalledTimes(1);
+		});
+	});
+});

packages/core/src/execution-engine/active-workflow-triggers.ts

@@ -2,7 +2,6 @@ import { Logger } from '@n8n/backend-common';
 import { Service } from '@n8n/di';
 import type {
 	INode,
-	IPollFunctions,
 	ITriggerResponse,
 	IWorkflowExecuteAdditionalData,
 	TriggerTime,
@@ -20,9 +19,9 @@ import {
 } from 'n8n-workflow';
 
 import { ErrorReporter } from '@/errors/error-reporter';
-import { SpanStatus, Tracing } from '@/observability';
 
 import type { IGetExecutePollFunctions, IGetExecuteTriggerFunctions } from './interfaces';
+import { PollTriggerExecutor } from './poll-trigger-executor';
 import { ScheduledTaskManager, type ScheduledTaskGroup } from './scheduled-task-manager';
 import { TriggersAndPollers } from './triggers-and-pollers';
 import {
@@ -49,7 +48,7 @@ export class ActiveWorkflowTriggers {
 		private readonly scheduledTaskManager: ScheduledTaskManager,
 		private readonly triggersAndPollers: TriggersAndPollers,
 		private readonly errorReporter: ErrorReporter,
-		private readonly tracing: Tracing,
+		private readonly pollTriggerExecutor: PollTriggerExecutor,
 	) {
 		this.logger = logger.scoped('workflow-publication');
 	}
@@ -356,12 +355,16 @@ export class ActiveWorkflowTriggers {
 
 		// Get all the trigger times
 		const cronExpressions = (pollTimes.item || []).map(toCronExpression);
-		const executePollTrigger = this.createPollTriggerExecuteFn(
-			workflowId,
+
+		// Capture this node activation's generation; removing or replacing the node
+		// invalidates only this poller, while leaving other workflow triggers intact.
+		const isCurrent = () =>
+			this.activeTriggersByWorkflowId.get(workflowId)?.isCurrent(node.id, token) ?? false;
+		const executePollTrigger = this.pollTriggerExecutor.create(
 			workflow,
 			node,
 			pollFunctions,
-			token,
+			isCurrent,
 		);
 
 		// Execute the poll trigger directly to be able to know if it works.
@@ -473,115 +476,6 @@ export class ActiveWorkflowTriggers {
 		}
 	}
 
-	/**
-	 * Creates a function that executes the poll() implementation for a poll
-	 * trigger node and triggers a workflow execution based on the output.
-	 */
-	private createPollTriggerExecuteFn(
-		workflowId: string,
-		workflow: Workflow,
-		node: INode,
-		pollFunctions: IPollFunctions,
-		token: TriggerRegistrationToken,
-	): (testingTrigger?: boolean) => Promise<void> {
-		// Capture this node activation's generation; removing or replacing the node
-		// invalidates only this poller, while leaving other workflow triggers intact.
-		const isCurrent = () =>
-			this.activeTriggersByWorkflowId.get(workflowId)?.isCurrent(node.id, token) ?? false;
-
-		return async (testingTrigger = false) => {
-			return await this.tracing.startSpan(
-				{
-					name: 'Workflow Trigger Poll',
-					op: 'trigger.poll',
-					attributes: {
-						...this.tracing.pickWorkflowAttributes(workflow),
-						...this.tracing.pickNodeAttributes(node),
-					},
-				},
-				async (span) => {
-					this.logger.debug(`Poll trigger initiated for workflow "${workflow.name}"`, {
-						workflowName: workflow.name,
-						workflowId: workflow.id,
-					});
-
-					// The initial activation poll runs inside ActiveWorkflowManager's
-					// outer acquireIsolate window, which also covers countTriggers
-					// afterwards. Acquiring here would release the outer bridge early
-					// (acquire is idempotent per caller; release deletes it). Scheduled
-					// polls fire from the cron scheduler's own async context outside
-					// that window and must acquire/release per tick — see CAT-3147.
-					const ownsIsolate = !testingTrigger;
-
-					// A scheduled poll can finish after the workflow was removed or
-					// reactivated, so drop it if superseded to prevent executing the old version.
-					if (!testingTrigger && !isCurrent()) {
-						this.logger.debug(`Skipping poll for superseded workflow "${workflow.name}"`, {
-							workflowId: workflow.id,
-						});
-						span.setStatus({ code: SpanStatus.ok });
-						return;
-					}
-
-					try {
-						if (ownsIsolate) await workflow.expression.acquireIsolate();
-
-						const pollResponse = await this.triggersAndPollers.runPollFunction(
-							workflow,
-							node,
-							pollFunctions,
-						);
-
-						// Same as the above `isCurrent` check; last chance to check before
-						// potentially starting the execution. Emitting now if superseded would run
-						// an execution against the old version of the workflow, so drop it.
-						// Bailing out here is safe even though `poll()` may have already advanced
-						// its state in the in-memory static data: persistence only happens inside
-						// `__emit` (`saveStaticData`), so the dropped call leaves the stored state
-						// untouched and the newly registered poller re-fetches the same events.
-						if (!testingTrigger && !isCurrent()) {
-							this.logger.debug(
-								`Discarding in-flight poll result for superseded workflow "${workflow.name}"`,
-								{ workflowId: workflow.id },
-							);
-							span.setStatus({ code: SpanStatus.ok });
-							return;
-						}
-
-						if (pollResponse !== null) {
-							pollFunctions.__emit(pollResponse);
-						}
-
-						span.setStatus({ code: SpanStatus.ok });
-					} catch (error) {
-						// If the poll trigger fails in the first activation
-						// throw the error back so we let the user know there is
-						// an issue with the trigger.
-						if (testingTrigger) {
-							span.setStatus({ code: SpanStatus.error });
-							throw error;
-						}
-
-						// Ignore poll errors that are against a superseded workflow
-						if (!isCurrent()) {
-							this.logger.debug(
-								`Ignoring in-flight poll error for superseded workflow "${workflow.name}"`,
-								{ workflowId: workflow.id },
-							);
-							span.setStatus({ code: SpanStatus.ok });
-							return;
-						}
-
-						span.setStatus({ code: SpanStatus.error });
-						pollFunctions.__emitError(error as Error);
-					} finally {
-						if (ownsIsolate) await workflow.expression.releaseIsolate();
-					}
-				},
-			);
-		};
-	}
-
 	private getOrCreateWorkflowTriggersState(workflowId: string) {
 		const existing = this.activeTriggersByWorkflowId.get(workflowId);
 		if (existing) return existing;

packages/core/src/execution-engine/index.ts

@@ -98,6 +98,7 @@ export type * from './interfaces';
 export * from './node-execution-context';
 export * from './node-execution-context/utils/execution-metadata';
 export * from './partial-execution-utils';
+export { PollTriggerExecutor } from './poll-trigger-executor';
 export { isEngineRequest } from './requests-response';
 export * from './routing-node';
 export * from './scheduled-task-manager';

packages/core/src/execution-engine/poll-trigger-executor.ts

@@ -0,0 +1,134 @@
+import { Logger } from '@n8n/backend-common';
+import { Service } from '@n8n/di';
+import type { INode, IPollFunctions, Workflow } from 'n8n-workflow';
+
+import { SpanStatus, Tracing } from '@/observability';
+
+import { TriggersAndPollers } from './triggers-and-pollers';
+
+/** Runs a poll trigger's `poll()` once. `testingTrigger` flags the initial activation poll. */
+export type PollTriggerExecuteFn = (testingTrigger?: boolean) => Promise<void>;
+
+/**
+ * Builds the runtime function that executes a poll trigger node. Kept separate
+ * from {@link ActiveWorkflowTriggers} so poll-trigger runtime logs are emitted
+ * under their own scope rather than the publication scope.
+ */
+@Service()
+export class PollTriggerExecutor {
+	constructor(
+		private readonly logger: Logger,
+		private readonly triggersAndPollers: TriggersAndPollers,
+		private readonly tracing: Tracing,
+	) {
+		this.logger = logger.scoped('poll-trigger');
+	}
+
+	/**
+	 * Creates a function that executes the poll() implementation for a poll
+	 * trigger node and triggers a workflow execution based on the output.
+	 *
+	 * @param isCurrent Tells whether this poller is still the active registration
+	 *   for its node. A scheduled poll can finish after the workflow was removed
+	 *   or reactivated, so a superseded result is dropped to avoid executing the
+	 *   old version.
+	 */
+	create(
+		workflow: Workflow,
+		node: INode,
+		pollFunctions: IPollFunctions,
+		isCurrent: () => boolean,
+	): PollTriggerExecuteFn {
+		return async (testingTrigger = false) => {
+			return await this.tracing.startSpan(
+				{
+					name: 'Workflow Trigger Poll',
+					op: 'trigger.poll',
+					attributes: {
+						...this.tracing.pickWorkflowAttributes(workflow),
+						...this.tracing.pickNodeAttributes(node),
+					},
+				},
+				async (span) => {
+					this.logger.debug(`Poll trigger initiated for workflow "${workflow.name}"`, {
+						workflowName: workflow.name,
+						workflowId: workflow.id,
+					});
+
+					// The initial activation poll runs inside ActiveWorkflowManager's
+					// outer acquireIsolate window, which also covers countTriggers
+					// afterwards. Acquiring here would release the outer bridge early
+					// (acquire is idempotent per caller; release deletes it). Scheduled
+					// polls fire from the cron scheduler's own async context outside
+					// that window and must acquire/release per tick — see CAT-3147.
+					const ownsIsolate = !testingTrigger;
+
+					// A scheduled poll can finish after the workflow was removed or
+					// reactivated, so drop it if superseded to prevent executing the old version.
+					if (!testingTrigger && !isCurrent()) {
+						this.logger.debug(`Skipping poll for superseded workflow "${workflow.name}"`, {
+							workflowId: workflow.id,
+						});
+						span.setStatus({ code: SpanStatus.ok });
+						return;
+					}
+
+					try {
+						if (ownsIsolate) await workflow.expression.acquireIsolate();
+
+						const pollResponse = await this.triggersAndPollers.runPollFunction(
+							workflow,
+							node,
+							pollFunctions,
+						);
+
+						// Same as the above `isCurrent` check; last chance to check before
+						// potentially starting the execution. Emitting now if superseded would run
+						// an execution against the old version of the workflow, so drop it.
+						// Bailing out here is safe even though `poll()` may have already advanced
+						// its state in the in-memory static data: persistence only happens inside
+						// `__emit` (`saveStaticData`), so the dropped call leaves the stored state
+						// untouched and the newly registered poller re-fetches the same events.
+						if (!testingTrigger && !isCurrent()) {
+							this.logger.debug(
+								`Discarding in-flight poll result for superseded workflow "${workflow.name}"`,
+								{ workflowId: workflow.id },
+							);
+							span.setStatus({ code: SpanStatus.ok });
+							return;
+						}
+
+						if (pollResponse !== null) {
+							pollFunctions.__emit(pollResponse);
+						}
+
+						span.setStatus({ code: SpanStatus.ok });
+					} catch (error) {
+						// If the poll trigger fails in the first activation
+						// throw the error back so we let the user know there is
+						// an issue with the trigger.
+						if (testingTrigger) {
+							span.setStatus({ code: SpanStatus.error });
+							throw error;
+						}
+
+						// Ignore poll errors that are against a superseded workflow
+						if (!isCurrent()) {
+							this.logger.debug(
+								`Ignoring in-flight poll error for superseded workflow "${workflow.name}"`,
+								{ workflowId: workflow.id },
+							);
+							span.setStatus({ code: SpanStatus.ok });
+							return;
+						}
+
+						span.setStatus({ code: SpanStatus.error });
+						pollFunctions.__emitError(error as Error);
+					} finally {
+						if (ownsIsolate) await workflow.expression.releaseIsolate();
+					}
+				},
+			);
+		};
+	}
+}