From cd21e0f2464c1adca0ac367cf80067f9cbbacdca Mon Sep 17 00:00:00 2001 From: Tomi Turtiainen <10324676+tomi@users.noreply.github.com> Date: Thu, 25 Jun 2026 15:59:23 +0300 Subject: [PATCH] refactor(core): Give poll trigger execution its own scoped logger (#32986) --- .../@n8n/config/src/configs/logging.config.ts | 1 + .../__tests__/active-workflow-manager.test.ts | 32 ++-- .../active-workflow-triggers.test.ts | 5 +- .../__tests__/poll-trigger-executor.test.ts | 146 ++++++++++++++++++ .../active-workflow-triggers.ts | 124 ++------------- packages/core/src/execution-engine/index.ts | 1 + .../execution-engine/poll-trigger-executor.ts | 134 ++++++++++++++++ 7 files changed, 311 insertions(+), 132 deletions(-) create mode 100644 packages/core/src/execution-engine/__tests__/poll-trigger-executor.test.ts create mode 100644 packages/core/src/execution-engine/poll-trigger-executor.ts diff --git a/packages/@n8n/config/src/configs/logging.config.ts b/packages/@n8n/config/src/configs/logging.config.ts index e714ed29dd8..a1cb4148129 100644 --- a/packages/@n8n/config/src/configs/logging.config.ts +++ b/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; diff --git a/packages/cli/src/__tests__/active-workflow-manager.test.ts b/packages/cli/src/__tests__/active-workflow-manager.test.ts index 3e21cf9cbc8..ae6371c1b5a 100644 --- a/packages/cli/src/__tests__/active-workflow-manager.test.ts +++ b/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({ id: 'wf-1', name: 'Test Workflow' }); const additionalData = mock(); + const logger = mock({ scoped: jest.fn().mockReturnValue(mock()) }); + const triggersAndPollers = { + runPollFunction: async (wf: Workflow, node: INode, pollFunctions: IPollFunctions) => + await wf.nodeTypes + .getByNameAndVersion(node.type, node.typeVersion) + .poll!.call(pollFunctions), + } as ConstructorParameters[2]; const realActiveWorkflowTriggers = new ActiveWorkflowTriggers( - mock({ scoped: jest.fn().mockReturnValue(mock()) }), + logger, scheduledTaskManager, - { - runPollFunction: async (wf: Workflow, node: INode, pollFunctions: IPollFunctions) => - await wf.nodeTypes - .getByNameAndVersion(node.type, node.typeVersion) - .poll!.call(pollFunctions), - } as ConstructorParameters[2], + triggersAndPollers, mock(), - { - startSpan: async (_options: unknown, fn: (span: unknown) => Promise) => - await fn({ setStatus: () => {} }), - pickWorkflowAttributes: () => ({}), - pickNodeAttributes: () => ({}), - } as unknown as ConstructorParameters[4], + new PollTriggerExecutor(logger, triggersAndPollers, new Tracing()), ); await realActiveWorkflowTriggers.addAllTriggers( @@ -758,7 +760,7 @@ describe('ActiveWorkflowManager', () => { realScheduledTaskManager, mock(), mock(), - mock(), + mock(), ); activeWorkflowManager = new ActiveWorkflowManager( mockLogger(), diff --git a/packages/core/src/execution-engine/__tests__/active-workflow-triggers.test.ts b/packages/core/src/execution-engine/__tests__/active-workflow-triggers.test.ts index 0ae212cfa43..ed24c6de756 100644 --- a/packages/core/src/execution-engine/__tests__/active-workflow-triggers.test.ts +++ b/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), ); }); diff --git a/packages/core/src/execution-engine/__tests__/poll-trigger-executor.test.ts b/packages/core/src/execution-engine/__tests__/poll-trigger-executor.test.ts new file mode 100644 index 00000000000..e4345297a7d --- /dev/null +++ b/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.scoped.mockReturnValue(logger); + + const triggersAndPollers = mock(); + const node = mock({ id: 'poll-node', name: 'Poll Node' }); + const pollFunctions = mock(); + + 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({ 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); + }); + }); +}); diff --git a/packages/core/src/execution-engine/active-workflow-triggers.ts b/packages/core/src/execution-engine/active-workflow-triggers.ts index c4fc2f998de..b1d84b8fc72 100644 --- a/packages/core/src/execution-engine/active-workflow-triggers.ts +++ b/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 { - // 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; diff --git a/packages/core/src/execution-engine/index.ts b/packages/core/src/execution-engine/index.ts index c45870773fa..dd3b9427bbe 100644 --- a/packages/core/src/execution-engine/index.ts +++ b/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'; diff --git a/packages/core/src/execution-engine/poll-trigger-executor.ts b/packages/core/src/execution-engine/poll-trigger-executor.ts new file mode 100644 index 00000000000..d879c354635 --- /dev/null +++ b/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; + +/** + * 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(); + } + }, + ); + }; + } +}