github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 05:29:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

refactor(core): Simplify the poll trigger task handler (no-changelog) (#40268)

Browse Source
This commit is contained in:
Lorent Lempereur 2026-10-06 09:07:33 +00:00 committed by GitHub
parent f26d7fbfe7
commit 39abed9124
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 51 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts
View File

@ -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<null>((_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<TriggersAndPollers>({ 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();
});

99
packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts
View File

@ -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<typeof poll>;
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<void> {
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);