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

feat(core): Bound a poll by the timeout of its occurrence (no-changelog) (#40518)
Some checks are pending
Build: Benchmark Image / build (push) Waiting to run
Details
CI: Master (Build, Test, Lint) / Build & format (push) Waiting to run
Details
CI: Master (Build, Test, Lint) / Unit tests (24.18.1) (push) Waiting to run
Details
CI: Master (Build, Test, Lint) / Unit tests (26.5.1) (push) Waiting to run
Details
CI: Master (Build, Test, Lint) / Lint (push) Waiting to run
Details
CI: Master (Build, Test, Lint) / Verify single-instance deps (npm-install) (push) Waiting to run
Details
CI: Master (Build, Test, Lint) / Notify Slack on failure (push) Blocked by required conditions
Details

Browse Source 
Co-authored-by: Veljko Terzic <veljko.terzic@n8n.io>
This commit is contained in:
Lorent Lempereur 2026-10-12 11:31:52 +00:00 committed by GitHub
parent 16e29da52f
commit 3dd805d9ed
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 235 additions and 201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2,7 +2,7 @@
import type { Logger } from '@n8n/backend-common';
import type { EventService, WorkflowSharingService } from '@n8n/backend-services';
import { mockLogger } from '@n8n/backend-test-utils';
import type { GlobalConfig, WorkflowsConfig } from '@n8n/config';
import type { WorkflowsConfig } from '@n8n/config';
import type { Project, WorkflowEntity, WorkflowHistory, WorkflowRepository } from '@n8n/db';
import type { UpdateResult } from '@n8n/typeorm';
import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
@ -890,7 +890,6 @@ describe('ActiveWorkflowManager', () => {
ownershipService,
mock(), // nodeTypes
pollCursorService,
mock<GlobalConfig>({ scheduler: { pollTimeoutSeconds: 45, leaseDurationSeconds: 60 } }),
mock(), // engineV2ActiveTriggers,
mock(), // workflowPublisherService
);

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

@ -1,11 +1,12 @@
/* eslint-disable @typescript-eslint/unbound-method */
import type { Logger } from '@n8n/backend-common';
import type { EventService } from '@n8n/backend-services';
import type { GlobalConfig } from '@n8n/config';
import type { PollerFullState, WorkflowRepository } from '@n8n/db';
import {
createDispatchReporter,
createScheduler,
LeaseLostError,
TaskTimeoutError,
type ClaimedTask,
type SchedulerMetrics,
type SchedulerTaskStore,
@ -39,8 +40,6 @@ describe('PollTriggerTaskHandler', () => {
const rootLogger = mock<Logger>({ scoped: vi.fn().mockReturnValue(scopedLogger) });
const eventService = mock<EventService>();
const pollTimeoutSeconds = 60;
const globalConfig = mock<GlobalConfig>({ scheduler: { pollTimeoutSeconds } });
const handler = new PollTriggerTaskHandler(
rootLogger,
@ -50,7 +49,6 @@ describe('PollTriggerTaskHandler', () => {
errorReporter,
pollBackoffService,
eventService,
globalConfig,
);
const onDispatch = vi.fn();
@ -104,9 +102,9 @@ describe('PollTriggerTaskHandler', () => {
scheduledFor,
runAt: scheduledFor,
status: 'running',
timeoutSeconds: 45,
attempts: 0,
maxAttempts: 1,
timeoutSeconds: 45,
leaseEpoch: 1,
...overrides,
});
@ -199,7 +197,11 @@ describe('PollTriggerTaskHandler', () => {
expect(triggerExecutionContextFactory.createPollExecutionContext).toHaveBeenCalledWith(
buildWorkflowData(),
triggerNode,
{ taskId: 'task-1', leaseEpoch: 1 },
{
fence: { taskId: 'task-1', leaseEpoch: 1 },
timeoutSeconds: 45,
remainingMs: expect.any(Function),
},
undefined,
);
expect(triggersAndPollers.runPollFunction).toHaveBeenCalledWith(
@ -221,7 +223,11 @@ describe('PollTriggerTaskHandler', () => {
expect(triggerExecutionContextFactory.createPollExecutionContext).toHaveBeenCalledWith(
buildWorkflowData(),
triggerNode,
{ taskId: 'task-1', leaseEpoch: 1 },
{
fence: { taskId: 'task-1', leaseEpoch: 1 },
timeoutSeconds: 45,
remainingMs: expect.any(Function),
},
{ lastItemId: 'prefetched' },
);
});
@ -487,81 +493,17 @@ describe('PollTriggerTaskHandler', () => {
});
describe('poll timeout', () => {
const pollTimeoutMs = pollTimeoutSeconds * 1000;
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
test.each(['poll', 'active-state lookup', 'failure write'])(
'rejects a timed-out poll if the lease was lost during the %s',
async (abortDuring) => {
const lease = new AbortController();
const reason = new Error('lease expired');
triggersAndPollers.runPollFunction.mockImplementationOnce(async () => {
if (abortDuring === 'poll') lease.abort(reason);
return await new Promise(() => {});
});
if (abortDuring === 'active-state lookup') {
workflowRepository.isActive.mockImplementationOnce(async () => {
lease.abort(reason);
return true;
});
}
if (abortDuring === 'failure write') {
pollBackoffService.recordFailure.mockImplementationOnce(async () => {
lease.abort(reason);
});
}
const rejected = expect(
handler.execute(buildTask(), report, runOf(lease.signal)),
).rejects.toBe(reason);
await vi.advanceTimersByTimeAsync(pollTimeoutMs);
await rejected;
expect(pollBackoffService.recordFailure).toHaveBeenCalledTimes(
abortDuring === 'failure write' ? 1 : 0,
);
expect(pollBackoffService.recordSuccess).not.toHaveBeenCalled();
expect(pollFunctions.__emit).not.toHaveBeenCalled();
expect(pollFunctions.__emitError).not.toHaveBeenCalled();
expect(pollFunctions.__commitCursor).not.toHaveBeenCalled();
expect(onDispatch).not.toHaveBeenCalled();
expect(releaseIsolate).toHaveBeenCalledTimes(1);
},
);
test('abandons a hanging poll as soon as the lease is lost', async () => {
const lease = new AbortController();
const reason = new Error('lease expired');
test('records a poll failure and reports no dispatch when the occurrence reaches its timeout', async () => {
const run = new AbortController();
triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {}));
const rejected = expect(
handler.execute(buildTask(), report, runOf(lease.signal)),
).rejects.toBe(reason);
await vi.advanceTimersByTimeAsync(1);
lease.abort(reason);
await rejected;
const settled = expect(handler.execute(buildTask(), report, runOf(run.signal))).resolves.toBe(
report.notDispatched(),
);
await vi.waitFor(() => expect(triggersAndPollers.runPollFunction).toHaveBeenCalled());
run.abort(new TaskTimeoutError(45));
await settled;
expect(eventService.emit).not.toHaveBeenCalledWith('poll-tick-timed-out', expect.anything());
expect(pollBackoffService.recordFailure).not.toHaveBeenCalled();
expect(pollFunctions.__emitError).not.toHaveBeenCalled();
expect(onDispatch).not.toHaveBeenCalled();
expect(releaseIsolate).toHaveBeenCalledTimes(1);
});
test('abandons a poll that outlives the timeout and reports no dispatch', async () => {
triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {}));
const executing = handler.execute(buildTask(), report, runOf(leaseSignal));
await vi.advanceTimersByTimeAsync(pollTimeoutMs);
await expect(executing).resolves.toBeDefined();
// Writes nothing: no cursor advance via __emit, and no error workflow run
// either, so the next occurrence covers the same poll window.
expect(pollFunctions.__emit).not.toHaveBeenCalled();
@ -571,6 +513,10 @@ describe('PollTriggerTaskHandler', () => {
expect(eventService.emit).toHaveBeenCalledWith('poll-tick-timed-out', {
nodeType: triggerNode.type,
});
expect(scopedLogger.warn).toHaveBeenCalledWith(
'Poll exceeded its timeout and was abandoned',
expect.objectContaining({ workflowId: 'wf-1', nodeId: 'node-1' }),
);
// The timeout counts as a transient poll failure, so a source that keeps
// hanging backs off like any failing source.
expect(pollBackoffService.recordFailure).toHaveBeenCalledWith(
@ -584,38 +530,41 @@ describe('PollTriggerTaskHandler', () => {
});
test('records no failure for a workflow deactivated during a timed-out poll', async () => {
const run = new AbortController();
workflowRepository.isActive.mockResolvedValue(false);
triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {}));
const executing = handler.execute(buildTask(), report, runOf(leaseSignal));
await vi.advanceTimersByTimeAsync(pollTimeoutMs);
await executing;
const settled = expect(handler.execute(buildTask(), report, runOf(run.signal))).resolves.toBe(
report.notDispatched(),
);
await vi.waitFor(() => expect(triggersAndPollers.runPollFunction).toHaveBeenCalled());
run.abort(new TaskTimeoutError(45));
await settled;
expect(pollBackoffService.recordFailure).not.toHaveBeenCalled();
});
test('keeps a poll that finishes just inside the timeout', async () => {
let resolvePoll: (data: INodeExecutionData[][]) => void = () => {};
triggersAndPollers.runPollFunction.mockReturnValue(
new Promise((resolve) => {
resolvePoll = resolve;
}),
test('abandons a hanging poll as soon as the lease is lost, and records nothing', async () => {
const run = new AbortController();
const reason = new LeaseLostError();
triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {}));
const rejected = expect(handler.execute(buildTask(), report, runOf(run.signal))).rejects.toBe(
reason,
);
await vi.waitFor(() => expect(triggersAndPollers.runPollFunction).toHaveBeenCalled());
run.abort(reason);
await rejected;
const executing = handler.execute(buildTask(), report, runOf(leaseSignal));
await vi.advanceTimersByTimeAsync(pollTimeoutMs - 1);
resolvePoll(pollData);
await executing;
expect(pollFunctions.__emit).toHaveBeenCalledWith(pollData);
expect(onDispatch).toHaveBeenCalledTimes(1);
expect(eventService.emit).not.toHaveBeenCalledWith('poll-tick-timed-out', expect.anything());
expect(pollBackoffService.recordFailure).not.toHaveBeenCalled();
// The deadline is cleared once the poll wins, so it can't outlive the tick.
expect(vi.getTimerCount()).toBe(0);
expect(pollFunctions.__emitError).not.toHaveBeenCalled();
expect(onDispatch).not.toHaveBeenCalled();
expect(releaseIsolate).toHaveBeenCalledTimes(1);
});
test('discards the data of an abandoned poll that resolves after the timeout', async () => {
const run = new AbortController();
let resolvePoll: (data: INodeExecutionData[][]) => void = () => {};
triggersAndPollers.runPollFunction.mockReturnValue(
new Promise((resolve) => {
@ -623,44 +572,56 @@ describe('PollTriggerTaskHandler', () => {
}),
);
const executing = handler.execute(buildTask(), report, runOf(leaseSignal));
await vi.advanceTimersByTimeAsync(pollTimeoutMs);
await executing;
const settled = expect(handler.execute(buildTask(), report, runOf(run.signal))).resolves.toBe(
report.notDispatched(),
);
await vi.waitFor(() => expect(triggersAndPollers.runPollFunction).toHaveBeenCalled());
run.abort(new TaskTimeoutError(45));
await settled;
resolvePoll(pollData);
await vi.advanceTimersByTimeAsync(0);
await new Promise((resolve) => setImmediate(resolve));
// The tick was already reported as abandoned, so the late data is dropped:
// no hand-off, no cursor advance, no dispatch.
// The tick was already abandoned, so the late data is dropped: no hand-off,
// no cursor advance, no dispatch.
expect(pollFunctions.__emit).not.toHaveBeenCalled();
expect(onDispatch).not.toHaveBeenCalled();
});
test('discards an abandoned poll that fails after the timeout', async () => {
const run = new AbortController();
let rejectPoll: (error: Error) => void = () => {};
const pollPromise = new Promise<null>((_resolve, reject) => {
rejectPoll = reject;
});
const runPollFunction = vi.fn();
// 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 }),
mock<TriggersAndPollers>({
runPollFunction: async () => {
runPollFunction();
return await pollPromise;
},
}),
workflowRepository,
errorReporter,
pollBackoffService,
eventService,
globalConfig,
);
const executing = plainHandler.execute(buildTask(), report, runOf(leaseSignal));
await vi.advanceTimersByTimeAsync(pollTimeoutMs);
await executing;
const settled = expect(
plainHandler.execute(buildTask(), report, runOf(run.signal)),
).resolves.toBe(report.notDispatched());
await vi.waitFor(() => expect(runPollFunction).toHaveBeenCalled());
run.abort(new TaskTimeoutError(45));
await settled;
rejectPoll(new Error('poll source unreachable'));
await vi.advanceTimersByTimeAsync(0);
await new Promise((resolve) => setImmediate(resolve));
// The tick was already reported as abandoned, so the late failure is dropped
// rather than routed to the error workflow.
// The tick was already 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();
@ -729,6 +690,56 @@ describe('PollTriggerTaskHandler', () => {
);
});
describe('timeout through the executor', () => {
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
// A retry would poll the hanging source again, and does not wait for backoff when it is off.
test('records a poll failure and completes the occurrence without a retry when a poll reaches its timeout', async () => {
const timeoutSeconds = 30;
const task = buildTask({ maxAttempts: 3, timeoutSeconds, runAt: new Date() });
const store = mock<SchedulerTaskStore>();
const metrics = mock<SchedulerMetrics>();
store.claimDueTasks.mockResolvedValue([task]);
store.beginDispatch.mockResolvedValue(1);
store.renewLease.mockResolvedValue(true);
store.rescheduleTask.mockResolvedValue(1);
store.markDispatched.mockResolvedValue(1);
store.completeTask.mockResolvedValue(1);
triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {}));
const scheduler = createScheduler({
hostId: 'test-host',
taskStore: store,
materializerTransaction: vi.fn(),
executor: { leaseSeconds: 60, lookaheadSeconds: 1 },
metrics,
});
scheduler.registerTaskHandler(POLL_TRIGGER_TASK_TYPE, handler);
await scheduler.execute();
await vi.advanceTimersByTimeAsync(timeoutSeconds * 1_000 + 1);
await vi.waitFor(() => expect(store.completeTask).toHaveBeenCalledTimes(1));
await scheduler.stop();
expect(metrics.recordTaskTimeout).toHaveBeenCalledExactlyOnceWith(POLL_TRIGGER_TASK_TYPE);
expect(pollBackoffService.recordFailure).toHaveBeenCalledWith(
expect.objectContaining({
error: expect.objectContaining({ failure: { cause: 'temporarily-unavailable' } }),
}),
);
expect(store.rescheduleTask).not.toHaveBeenCalled();
expect(store.failTaskTerminal).not.toHaveBeenCalled();
expect(store.completeTask).toHaveBeenCalledTimes(1);
expect(pollFunctions.__emit).not.toHaveBeenCalled();
expect(pollFunctions.__emitError).not.toHaveBeenCalled();
});
});
describe('failures', () => {
test('rejects a task whose payload is missing workflowId or nodeId', async () => {
const task = buildTask({ payload: { nodeId: 'node-1' } });

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

@ -1,10 +1,9 @@
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 { TaskTimeoutError } from '@n8n/scheduler';
import type {
ClaimedTask,
DispatchDecision,
@ -51,8 +50,6 @@ class PollTimeoutError extends OperationalError {
export class PollTriggerTaskHandler implements TaskHandler {
readonly taskType = POLL_TRIGGER_TASK_TYPE;
private readonly pollTimeoutMs: number;
constructor(
private logger: Logger,
private readonly triggerExecutionContextFactory: TriggerExecutionContextFactory,
@ -61,16 +58,14 @@ export class PollTriggerTaskHandler implements TaskHandler {
private readonly errorReporter: ErrorReporter,
private readonly pollBackoffService: PollBackoffService,
private readonly eventService: EventService,
globalConfig: GlobalConfig,
) {
this.logger = this.logger.scoped('scheduler');
this.pollTimeoutMs = globalConfig.scheduler.pollTimeoutSeconds * Time.seconds.toMilliseconds;
}
async execute(
task: ClaimedTask,
report: DispatchReporter,
{ signal: leaseSignal }: TaskRun,
{ signal: leaseSignal, remainingMs }: TaskRun,
): Promise<DispatchDecision> {
// 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.
@ -121,7 +116,11 @@ export class PollTriggerTaskHandler implements TaskHandler {
await this.triggerExecutionContextFactory.createPollExecutionContext(
workflowData,
node,
{ taskId: task.id, leaseEpoch: task.leaseEpoch },
{
fence: { taskId: task.id, leaseEpoch: task.leaseEpoch },
timeoutSeconds: task.timeoutSeconds,
remainingMs,
},
state?.cursor,
);
@ -141,43 +140,21 @@ export class PollTriggerTaskHandler implements TaskHandler {
let polled = false;
try {
leaseSignal.throwIfAborted();
// `poll()` takes no abort signal, so the deadline abandons it rather than
// cancelling it: the call keeps running until it settles on its own, and its
// outcome is discarded. The cursor never moves on that path (it only moves
// through the staged commit or __emit below), so an abandoned tick leaves
// the poll window untouched for the next occurrence to cover.
const deadline = new AbortController();
const deadlineTimer = setTimeout(() => deadline.abort(), this.pollTimeoutMs).unref();
const abandon = AbortSignal.any([leaseSignal, deadline.signal]);
// `poll()` takes no abort signal, so the timeout of the occurrence or a lost
// claim abandons it rather than cancelling it: the call keeps running until
// it settles on its own, and its outcome is discarded. The cursor never moves
// on that path (it only moves through the staged commit or __emit below), so
// an abandoned tick leaves the poll window untouched for the next occurrence.
const poll = this.triggersAndPollers.runPollFunction(workflow, node, pollFunctions);
let pollResponse: Awaited<typeof poll>;
try {
await Promise.race([poll, aborted(abandon, pollFunctions)]);
leaseSignal.throwIfAborted();
if (deadline.signal.aborted) {
this.eventService.emit('poll-tick-timed-out', { nodeType: node.type });
this.logger.warn('Poll exceeded its timeout and was abandoned', {
...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.
await this.recordFailureIfActive(
workflowId,
nodeId,
new PollTimeoutError(),
state,
leaseSignal,
);
leaseSignal.throwIfAborted();
return report.notDispatched();
}
pollResponse = await poll;
} finally {
clearTimeout(deadlineTimer);
await Promise.race([poll, aborted(leaseSignal, pollFunctions)]);
if (leaseSignal.reason instanceof TaskTimeoutError) {
await this.recordPollTimeout(logContext, node.type, state);
// A retry would poll the hanging source again, and backoff may be off.
// The next occurrence covers the same window, so this one completes.
return report.notDispatched();
}
leaseSignal.throwIfAborted();
const pollResponse = await poll;
polled = true;
await this.pollBackoffService.recordSuccess({ workflowId, nodeId, state });
@ -248,15 +225,31 @@ export class PollTriggerTaskHandler implements TaskHandler {
});
}
/**
* 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.
*/
private async recordPollTimeout(
logContext: { taskId: string; jobId: number; workflowId: string; nodeId: string },
nodeType: string,
state: PollerFailureState | null,
): Promise<void> {
this.eventService.emit('poll-tick-timed-out', { nodeType });
this.logger.warn('Poll exceeded its timeout and was abandoned', logContext);
const { workflowId, nodeId } = logContext;
await this.recordFailureIfActive(workflowId, nodeId, new PollTimeoutError(), state);
}
private async recordFailureIfActive(
workflowId: string,
nodeId: string,
error: unknown,
state: PollerFailureState | null,
leaseSignal: AbortSignal,
leaseSignal?: AbortSignal,
): Promise<void> {
const isActive = await this.workflowRepository.isActive(workflowId).catch(() => true);
leaseSignal.throwIfAborted();
leaseSignal?.throwIfAborted();
if (isActive) {
await this.pollBackoffService.recordFailure({
workflowId,

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

@ -1,7 +1,6 @@
/* eslint-disable @typescript-eslint/unbound-method */
import type { Logger } from '@n8n/backend-common';
import type { EventService } from '@n8n/backend-services';
import type { GlobalConfig } from '@n8n/config';
import type { Project, WorkflowEntity } from '@n8n/db';
import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
@ -119,7 +118,6 @@ describe('TriggerExecutionContextFactory', () => {
ownershipService,
nodeTypes,
pollCursorService,
mock<GlobalConfig>({ scheduler: { pollTimeoutSeconds: 45, leaseDurationSeconds: 60 } }),
engineV2ActiveTriggers,
workflowPublisherService,
);
@ -1341,14 +1339,16 @@ describe('TriggerExecutionContextFactory', () => {
});
describe('getPollBudgetMs', () => {
const NOW = 1_000_000;
beforeEach(() => {
vi.spyOn(performance, 'now').mockReturnValue(NOW);
});
const buildBudgetContext = (
pollTimeoutSeconds: number,
leaseDurationSeconds: number,
fence?: { taskId: string; leaseEpoch: number },
timeoutSeconds?: number,
deadline = NOW + (timeoutSeconds ?? 0) * 1_000,
) => {
const globalConfig = mock<GlobalConfig>({
scheduler: { pollTimeoutSeconds, leaseDurationSeconds },
});
const budgetFactory = new TriggerExecutionContextFactory(
mock<Logger>({ scoped: vi.fn().mockReturnValue(scopedLogger) }),
errorReporter,
@ -1363,7 +1363,6 @@ describe('TriggerExecutionContextFactory', () => {
ownershipService,
nodeTypes,
pollCursorService,
globalConfig,
engineV2ActiveTriggers,
mock(),
);
@ -1373,34 +1372,50 @@ describe('TriggerExecutionContextFactory', () => {
mode,
activation,
async () => mock<IWorkflowBase>({ id: 'wf-1', name: 'Test Workflow' }),
fence,
timeoutSeconds === undefined
? undefined
: {
fence: { taskId: 'task-1', leaseEpoch: 1 },
timeoutSeconds,
remainingMs: () => deadline - performance.now(),
},
);
return getPollFunctions(workflow, node, additionalData, mode, activation);
};
const fence = { taskId: 'task-1', leaseEpoch: 1 };
// Budget = poll timeout minus a margin of max(20%, 5s), so a node
// that exhausts it still finishes its in-flight batch before the
// engine abandons the tick. The lease is renewed, so it does not cap it.
// Budget = the timeout of the occurrence minus a margin of max(20%, 5s), so
// a node that exhausts it still finishes its in-flight batch before the
// executor aborts the run. The lease is renewed, so it does not cap it.
test.each([
{ pollTimeoutSeconds: 45, leaseDurationSeconds: 60, expected: 36_000 },
{ pollTimeoutSeconds: 100, leaseDurationSeconds: 60, expected: 80_000 },
{ timeoutSeconds: 45, expected: 36_000 },
{ timeoutSeconds: 100, expected: 80_000 },
// The margin never eats more than half the ceiling, so a tiny (but
// schema-valid) timeout still yields a positive budget.
{ pollTimeoutSeconds: 4, leaseDurationSeconds: 60, expected: 2_000 },
{ timeoutSeconds: 4, expected: 2_000 },
])(
'derives $expected ms from a $pollTimeoutSeconds s timeout under a $leaseDurationSeconds s lease',
({ pollTimeoutSeconds, leaseDurationSeconds, expected }) => {
const budgetContext = buildBudgetContext(pollTimeoutSeconds, leaseDurationSeconds, fence);
'derives $expected ms from an occurrence timeout of $timeoutSeconds s',
({ timeoutSeconds, expected }) => {
const budgetContext = buildBudgetContext(timeoutSeconds);
expect(budgetContext.getPollBudgetMs()).toBe(expected);
},
);
test('subtracts the time that setup used before the poll', () => {
const budgetContext = buildBudgetContext(45, NOW + 45_000);
vi.mocked(performance.now).mockReturnValue(NOW + 10_000);
expect(budgetContext.getPollBudgetMs()).toBe(26_000);
});
test('returns 0 once setup has used the whole budget', () => {
const budgetContext = buildBudgetContext(45, NOW + 45_000);
vi.mocked(performance.now).mockReturnValue(NOW + 40_000);
expect(budgetContext.getPollBudgetMs()).toBe(0);
});
test('keeps the generous PollContext default for a poll that runs without a lease', () => {
// The legacy in-memory path has no poll timeout and no lease, so the
// The legacy in-memory path has no occurrence and no lease, so the
// scheduler-derived budget must not apply there.
const budgetContext = buildBudgetContext(45, 60);
const budgetContext = buildBudgetContext();
expect(budgetContext.getPollBudgetMs()).toBe(300_000);
});
});
@ -1667,7 +1682,7 @@ describe('TriggerExecutionContextFactory', () => {
mode,
activation,
async () => mock<IWorkflowBase>({ id: 'wf-1', name: 'Test Workflow' }),
fence,
{ fence, timeoutSeconds: 45, remainingMs: () => 45_000 },
);
const fencedContext = getPollFunctions(
workflow,
@ -1780,9 +1795,13 @@ describe('TriggerExecutionContextFactory', () => {
const getExecutePollFunctionsSpy = vi
.spyOn(factory, 'getExecutePollFunctions')
.mockReturnValue(getPollFunctions as unknown as IGetExecutePollFunctions);
const fence = { taskId: 'task-1', leaseEpoch: 3 };
const leasedPoll = {
fence: { taskId: 'task-1', leaseEpoch: 3 },
timeoutSeconds: 45,
remainingMs: () => 45_000,
};
await factory.createPollExecutionContext(workflowData, pollNode, fence);
await factory.createPollExecutionContext(workflowData, pollNode, leasedPoll);
expect(getExecutePollFunctionsSpy).toHaveBeenCalledWith(
workflowData,
@ -1790,7 +1809,7 @@ describe('TriggerExecutionContextFactory', () => {
'trigger',
'update',
expect.any(Function),
fence,
leasedPoll,
undefined,
);
});
@ -1805,10 +1824,14 @@ describe('TriggerExecutionContextFactory', () => {
const getExecutePollFunctionsSpy = vi
.spyOn(factory, 'getExecutePollFunctions')
.mockReturnValue(getPollFunctions as unknown as IGetExecutePollFunctions);
const fence = { taskId: 'task-1', leaseEpoch: 3 };
const leasedPoll = {
fence: { taskId: 'task-1', leaseEpoch: 3 },
timeoutSeconds: 45,
remainingMs: () => 45_000,
};
const prefetched = { lastItemId: 'prefetched' };
await factory.createPollExecutionContext(workflowData, pollNode, fence, prefetched);
await factory.createPollExecutionContext(workflowData, pollNode, leasedPoll, prefetched);
expect(getExecutePollFunctionsSpy).toHaveBeenCalledWith(
workflowData,
@ -1816,7 +1839,7 @@ describe('TriggerExecutionContextFactory', () => {
'trigger',
'update',
expect.any(Function),
fence,
leasedPoll,
prefetched,
);
});

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

@ -1,6 +1,5 @@
import { Logger } from '@n8n/backend-common';
import { EventService } from '@n8n/backend-services';
import { GlobalConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
import type { IWorkflowDb, PollerCursor, PollLeaseFence } from '@n8n/db';
import { Service } from '@n8n/di';
@ -69,6 +68,14 @@ export type TriggerFailureHandler = (opts: {
activation: WorkflowActivateMode;
}) => void;
/** A poll run by a claimed occurrence: the fence of its writes, and the timeout of the occurrence. */
export interface LeasedPoll {
fence: PollLeaseFence;
timeoutSeconds: number;
/** Milliseconds until the occurrence times out. */
remainingMs(): number;
}
/**
* Builds the execution-context functions (`IGetExecuteTriggerFunctions` /
* `IGetExecutePollFunctions`) that n8n-core uses to wire up active and poll
@ -95,7 +102,6 @@ export class TriggerExecutionContextFactory {
private readonly ownershipService: OwnershipService,
private readonly nodeTypes: NodeTypes,
private readonly pollCursorService: PollCursorService,
private readonly globalConfig: GlobalConfig,
private readonly engineV2ActiveTriggers: EngineV2ActiveTriggers,
private readonly workflowPublisherService: WorkflowPublisherService,
) {
@ -320,18 +326,20 @@ export class TriggerExecutionContextFactory {
// service (flag on). Once the feature flag is removed, we'll call the
// service directly and this parameter will go away.
resolveWorkflowData: () => Promise<IWorkflowBase>,
fence?: PollLeaseFence,
leasedPoll?: LeasedPoll,
prefetchedCursor?: PollerCursor,
): IGetExecutePollFunctions {
const fence = leasedPoll?.fence;
return (workflow: Workflow, node: INode) => {
// A poll must finish inside the handler's abandon deadline; past it, its
// commits are discarded. The lease is renewed while the poll runs, so it
// does not bound the poll. The margin — 20%, at least 5s, at most half the
// ceiling — leaves room for the trailing hand-off and cursor commit.
const ceilingMs =
this.globalConfig.scheduler.pollTimeoutSeconds * Time.seconds.toMilliseconds;
// A poll must finish inside the timeout of its occurrence; past it, the
// executor aborts the run and its commits are discarded. The lease is
// renewed while the poll runs, so it does not bound the poll. The margin
// — 20%, at least 5s, at most half the ceiling — leaves room for the
// trailing hand-off and cursor commit.
const ceilingMs = (leasedPoll?.timeoutSeconds ?? 0) * Time.seconds.toMilliseconds;
const marginMs = Math.min(Math.max(0.2 * ceilingMs, 5_000), ceilingMs / 2);
const pollBudgetMs = ceilingMs - marginMs;
// Setup before `poll()` uses part of the timeout, so the budget counts down with it.
const pollBudgetMs = () => Math.max(0, (leasedPoll?.remainingMs() ?? 0) - marginMs);
// A poll's staged snapshot lives in an async scope entered per poll, rather
// than in a variable per node: only the poll that staged it can commit it, and
// two overlapping polls of the same node never share a slot. An unmigrated
@ -489,7 +497,7 @@ export class TriggerExecutionContextFactory {
resolveNodeStaticData,
// Only a leased (durable) poll is bounded by the timeout; a
// legacy in-memory poll keeps PollContext's generous default.
fence ? () => pollBudgetMs : undefined,
fence ? pollBudgetMs : undefined,
);
};
}
@ -502,7 +510,7 @@ export class TriggerExecutionContextFactory {
async createPollExecutionContext(
workflowData: IWorkflowBase,
node: INode,
fence?: PollLeaseFence,
leasedPoll?: LeasedPoll,
prefetchedCursor?: PollerCursor,
): Promise<{ workflow: Workflow; pollFunctions: IPollFunctions }> {
const workflow = new Workflow({
@ -532,7 +540,7 @@ export class TriggerExecutionContextFactory {
'trigger',
'update',
resolveWorkflowData,
fence,
leasedPoll,
prefetchedCursor,
);
// getPollFunctions already closed over these; its signature still requires them.