diff --git a/packages/cli/src/__tests__/active-workflow-manager.test.ts b/packages/cli/src/__tests__/active-workflow-manager.test.ts index aa74f698494..86327295ebd 100644 --- a/packages/cli/src/__tests__/active-workflow-manager.test.ts +++ b/packages/cli/src/__tests__/active-workflow-manager.test.ts @@ -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({ scheduler: { pollTimeoutSeconds: 45, leaseDurationSeconds: 60 } }), mock(), // engineV2ActiveTriggers, mock(), // workflowPublisherService ); 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 5ac5186a053..6fc37e695c6 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 @@ -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({ scoped: vi.fn().mockReturnValue(scopedLogger) }); const eventService = mock(); - const pollTimeoutSeconds = 60; - const globalConfig = mock({ 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((_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({ runPollFunction: async () => await pollPromise }), + mock({ + 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(); + const metrics = mock(); + 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' } }); 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 39cce2c8822..2598aee252b 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 @@ -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 { // 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; - 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 { + 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 { const isActive = await this.workflowRepository.isActive(workflowId).catch(() => true); - leaseSignal.throwIfAborted(); + leaseSignal?.throwIfAborted(); if (isActive) { await this.pollBackoffService.recordFailure({ workflowId, diff --git a/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts b/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts index f0620a6d8b6..c53f0082410 100644 --- a/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts @@ -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({ 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({ - scheduler: { pollTimeoutSeconds, leaseDurationSeconds }, - }); const budgetFactory = new TriggerExecutionContextFactory( mock({ 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({ 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({ 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, ); }); diff --git a/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts b/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts index a8b9f5ab748..4ba0e7baa06 100644 --- a/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts +++ b/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts @@ -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, - 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.