diff --git a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts index c6ea51207c2..9297797ea0c 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts @@ -477,13 +477,23 @@ describe('PollTriggerTaskHandler', () => { test('discards an abandoned poll that fails after the timeout', async () => { let rejectPoll: (error: Error) => void = () => {}; - triggersAndPollers.runPollFunction.mockReturnValue( - new Promise((_resolve, reject) => { - rejectPoll = reject; - }), + const pollPromise = new Promise((_resolve, reject) => { + rejectPoll = reject; + }); + // A plain stub, because a vi.fn() subscribes to the promise it returns and so + // would mark the late rejection as handled. + const plainHandler = new PollTriggerTaskHandler( + rootLogger, + triggerExecutionContextFactory, + mock({ runPollFunction: async () => await pollPromise }), + workflowRepository, + errorReporter, + pollBackoffService, + eventService, + globalConfig, ); - const executing = handler.execute(buildTask(), report); + const executing = plainHandler.execute(buildTask(), report); await vi.advanceTimersByTimeAsync(pollTimeoutMs); await executing; rejectPoll(new Error('poll source unreachable')); @@ -491,6 +501,7 @@ describe('PollTriggerTaskHandler', () => { // The tick was already reported as abandoned, so the late failure is dropped // rather than routed to the error workflow. + expect(eventService.emit).toHaveBeenCalledWith('poll-tick-timed-out', expect.anything()); expect(pollFunctions.__emitError).not.toHaveBeenCalled(); expect(onDispatch).not.toHaveBeenCalled(); }); diff --git a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts index a454de7a559..9dc92b904ae 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts @@ -2,6 +2,7 @@ import { Logger } from '@n8n/backend-common'; import { EventService } from '@n8n/backend-services'; import { GlobalConfig } from '@n8n/config'; import { Time } from '@n8n/constants'; +import type { PollerFailureState } from '@n8n/db'; import { WorkflowRepository } from '@n8n/db'; import { Service } from '@n8n/di'; import type { ClaimedTask, DispatchDecision, DispatchReporter, TaskHandler } from '@n8n/scheduler'; @@ -76,15 +77,13 @@ export class PollTriggerTaskHandler implements TaskHandler { // A setup failure here retries to N8N_SCHEDULER_MAX_ATTEMPTS then dead-letters, // unlike a `poll()` runtime failure below, which routes to the error workflow instead. const { workflowId, nodeId } = this.parsePayload(task); + const logContext = { taskId: task.id, jobId: task.jobId, workflowId, nodeId }; const now = new Date(); const state = await this.pollBackoffService.getState(workflowId, nodeId).catch(() => null); if (this.pollBackoffService.isBackingOff(state, now)) { this.logger.debug('Poll is backing off; skipping this occurrence', { - taskId: task.id, - jobId: task.jobId, - workflowId, - nodeId, + ...logContext, backoffUntil: state?.backoffUntil, }); return report.notDispatched(); @@ -96,12 +95,7 @@ export class PollTriggerTaskHandler implements TaskHandler { ); if (workflowData === null) { - this.logger.debug('Workflow has no published version. Skipping the occurrence.', { - taskId: task.id, - jobId: task.jobId, - workflowId, - nodeId, - }); + this.logger.debug('Workflow has no published version. Skipping the occurrence.', logContext); return report.notDispatched(); } @@ -113,7 +107,7 @@ export class PollTriggerTaskHandler implements TaskHandler { if (!this.hasHealthyNodeIds(workflowData)) { this.logger.debug( 'Published version has duplicate or missing node ids; skipping the occurrence until it is healed', - { taskId: task.id, jobId: task.jobId, workflowId, nodeId }, + logContext, ); return report.notDispatched(); } @@ -155,9 +149,6 @@ export class PollTriggerTaskHandler implements TaskHandler { // the poll window untouched for the next occurrence to cover. const deadline = timeoutAfter(this.pollTimeoutMs); const poll = this.triggersAndPollers.runPollFunction(workflow, node, pollFunctions); - // Deliberately not chained: keeps an abandoned poll's eventual rejection from - // surfacing as an unhandled rejection once the race has moved on. - poll.catch(() => {}); let pollResponse: Awaited; try { @@ -165,25 +156,13 @@ export class PollTriggerTaskHandler implements TaskHandler { if (outcome === TIMED_OUT) { this.eventService.emit('poll-tick-timed-out', { nodeType: node.type }); this.logger.warn('Poll exceeded its timeout and was abandoned', { - taskId: task.id, - jobId: task.jobId, - workflowId, - nodeId, + ...logContext, pollTimeoutMs: this.pollTimeoutMs, }); // Not routed to the error workflow: an abandoned poll produces no run, and // an error run is one. It does count as a poll failure, so a source that // keeps hanging is re-polled at a widening interval like any failing source. - const isActive = await this.workflowRepository.isActive(workflowId).catch(() => true); - if (isActive) { - await this.pollBackoffService.recordFailure({ - workflowId, - nodeId, - error: new PollTimeoutError(), - state, - now: new Date(), // Fresh clock, not the tick's - }); - } + await this.recordFailureIfActive(workflowId, nodeId, new PollTimeoutError(), state); return report.notDispatched(); } pollResponse = outcome; @@ -199,23 +178,16 @@ export class PollTriggerTaskHandler implements TaskHandler { // the workflow may have been deactivated while it was in flight. There is // no in-memory registration to check here, so re-read the stored active state. if (!(await this.workflowRepository.isActive(workflowId))) { - this.logger.debug('Workflow deactivated during poll; discarding the result', { - taskId: task.id, - jobId: task.jobId, - workflowId, - nodeId, - }); + this.logger.debug( + 'Workflow deactivated during poll; discarding the result', + logContext, + ); return report.notDispatched(); } // __emit saves the cursor and starts the run without waiting on it. pollFunctions.__emit(pollResponse); - this.logger.debug('Poll returned new data; handed off to a new execution', { - taskId: task.id, - jobId: task.jobId, - workflowId, - nodeId, - }); + this.logger.debug('Poll returned new data; handed off to a new execution', logContext); return report.dispatched(); } @@ -229,44 +201,25 @@ export class PollTriggerTaskHandler implements TaskHandler { // The poll itself succeeded, so a failed cursor write is logged rather // than routed to the error workflow. this.errorReporter.error(error, { - extra: { taskId: task.id, jobId: task.jobId, workflowId, nodeId }, + extra: logContext, }); this.logger.error( 'Failed to commit the poll cursor; the next poll repeats the same window', - { taskId: task.id, jobId: task.jobId, workflowId, nodeId, error }, + { ...logContext, error }, ); } - this.logger.debug('Poll returned no new data; nothing to hand off', { - taskId: task.id, - jobId: task.jobId, - workflowId, - nodeId, - }); + this.logger.debug('Poll returned no new data; nothing to hand off', logContext); return report.notDispatched(); } catch (error) { // Routed to the error workflow instead of rethrown, which would retry and // dead-letter without ever running it. __emitError commits no cursor, so // the cursor holds and the next tick retries the same window. if (!polled) { - const isActive = await this.workflowRepository.isActive(workflowId).catch(() => true); - if (isActive) { - await this.pollBackoffService.recordFailure({ - workflowId, - nodeId, - error, - state, - now: new Date(), // Fresh clock, not the tick's - }); - } + await this.recordFailureIfActive(workflowId, nodeId, error, state); } pollFunctions.__emitError(ensureError(error)); - this.logger.debug('Poll failed at runtime; routed to the error workflow', { - taskId: task.id, - jobId: task.jobId, - workflowId, - nodeId, - }); + this.logger.debug('Poll failed at runtime; routed to the error workflow', logContext); // The error was handed off, so this occurrence is handled and must not retry. return report.dispatched(); } finally { @@ -275,6 +228,24 @@ export class PollTriggerTaskHandler implements TaskHandler { }); } + private async recordFailureIfActive( + workflowId: string, + nodeId: string, + error: unknown, + state: PollerFailureState | null, + ): Promise { + const isActive = await this.workflowRepository.isActive(workflowId).catch(() => true); + if (isActive) { + await this.pollBackoffService.recordFailure({ + workflowId, + nodeId, + error, + state, + now: new Date(), // Use the failure time, not the tick start. + }); + } + } + /** The invariant the activation-time healer guarantees: every node id unique and non-empty. */ private hasHealthyNodeIds(workflowData: IWorkflowBase): boolean { const ids = workflowData.nodes.map((node) => node.id);