From fdad09e67f4ffe324dcff076b37bd8ff647780ba Mon Sep 17 00:00:00 2001 From: Emilia <100027345+sovietspaceship@users.noreply.github.com> Date: Mon, 24 Aug 2026 15:33:04 +0000 Subject: [PATCH] feat(core): Bound a durable-scheduler poll tick with a configurable timeout (no-changelog) (#35874) Co-authored-by: Claude Opus 5 Co-authored-by: Lorent Lempereur --- .../__tests__/scheduler.config.test.ts | 12 ++ .../config/src/configs/scheduler.config.ts | 19 +++ packages/@n8n/config/test/config.test.ts | 1 + .../maps/poll-trigger-metrics.event-map.ts | 10 +- .../poll-trigger-metrics.service.test.ts | 18 +++ .../poll-trigger-metrics.service.ts | 16 ++- .../__tests__/durable-scheduler.test.ts | 80 ++++++++++++ .../cli/src/scheduling/durable-scheduler.ts | 21 +++ .../poll-trigger-task-handler.test.ts | 120 ++++++++++++++++++ .../poll-trigger-task-handler.ts | 87 +++++++++++-- 10 files changed, 370 insertions(+), 14 deletions(-) diff --git a/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts b/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts index 5339c967c3c..3a0fb3845b1 100644 --- a/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts +++ b/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts @@ -43,6 +43,7 @@ describe('SchedulerConfig', () => { expect(scheduler.allowSkipDurableScheduler).toBe(false); expect(scheduler.maxAttempts).toBe(5); expect(scheduler.enabledForPollTriggers).toBe(false); + expect(scheduler.pollTimeoutSeconds).toBe(45); }); }); @@ -73,6 +74,7 @@ describe('SchedulerConfig', () => { vi.stubEnv('N8N_SCHEDULER_MAX_CONCURRENT_PASSES', '4'); vi.stubEnv('N8N_SCHEDULER_MAX_ATTEMPTS', '3'); vi.stubEnv('N8N_SCHEDULER_POLL_TRIGGERS_ENABLED', 'true'); + vi.stubEnv('N8N_SCHEDULER_POLL_TIMEOUT', '30'); const { scheduler } = Container.get(GlobalConfig); @@ -93,6 +95,7 @@ describe('SchedulerConfig', () => { expect(scheduler.maxConcurrentPasses).toBe(4); expect(scheduler.maxAttempts).toBe(3); expect(scheduler.enabledForPollTriggers).toBe(true); + expect(scheduler.pollTimeoutSeconds).toBe(30); }); it('should expose the durable-scheduler skip escape hatch via env', () => { @@ -103,6 +106,15 @@ describe('SchedulerConfig', () => { expect(scheduler.allowSkipDurableScheduler).toBe(true); }); + it('should fall back to the default poll timeout when the value exceeds one day', () => { + vi.spyOn(console, 'warn').mockImplementation(() => {}); + vi.stubEnv('N8N_SCHEDULER_POLL_TIMEOUT', '86401'); + + const { scheduler } = Container.get(GlobalConfig); + + expect(scheduler.pollTimeoutSeconds).toBe(45); + }); + it('should allow disabling the min-interval clamp with 0', () => { vi.stubEnv('N8N_SCHEDULER_MIN_INTERVAL', '0'); diff --git a/packages/@n8n/config/src/configs/scheduler.config.ts b/packages/@n8n/config/src/configs/scheduler.config.ts index d389d00dd18..04c306d29b2 100644 --- a/packages/@n8n/config/src/configs/scheduler.config.ts +++ b/packages/@n8n/config/src/configs/scheduler.config.ts @@ -268,6 +268,25 @@ export class SchedulerConfig { @Env('N8N_SCHEDULER_POLL_TRIGGERS_ENABLED') enabledForPollTriggers: boolean = false; + /** + * How long, in seconds, a single poll of an external source (an inbox, an API) + * may take before it is abandoned. Defaults to 45 seconds. + * + * An abandoned poll skips no data: its position in the source is left where it + * was and the next scheduled poll covers the same ground. It counts as a poll + * failure, so a source that keeps timing out is polled at a widening interval. + * Guards against a poll stuck on an unresponsive source running indefinitely. + * + * Keep it below {@link leaseDurationSeconds}: the deadline only starts after + * the occurrence's setup reads, so a poll allowed to run as long as the claim + * on its run can still be in flight when that claim expires and another + * instance takes the run over. The default leaves that headroom, and the + * scheduler warns at startup when the timeout reaches the lease duration. + * Must be greater than 0 and at most one day. + */ + @Env('N8N_SCHEDULER_POLL_TIMEOUT', positiveIntSchema.max(Time.days.toSeconds)) + pollTimeoutSeconds: number = 45; + /** * Whether a poll trigger's cursor advance and the execution it produced are saved * together, atomically. When disabled, a crash between the two can leave a poll diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index e67ee196295..75c7ec6ba96 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/packages/@n8n/config/test/config.test.ts @@ -478,6 +478,7 @@ describe('GlobalConfig', () => { maxConcurrentPasses: 10, triggerNodeMode: 'legacy', enabledForPollTriggers: false, + pollTimeoutSeconds: 45, allowSkipDurableScheduler: false, maxAttempts: 5, misfireGraceSeconds: 60, diff --git a/packages/cli/src/events/maps/poll-trigger-metrics.event-map.ts b/packages/cli/src/events/maps/poll-trigger-metrics.event-map.ts index 3986393c9d8..cd4fff58737 100644 --- a/packages/cli/src/events/maps/poll-trigger-metrics.event-map.ts +++ b/packages/cli/src/events/maps/poll-trigger-metrics.event-map.ts @@ -1,8 +1,8 @@ /** * Low-cardinality metrics events emitted by the poll-cursor persistence path and - * consumed by the Prometheus poll-trigger collector. Payloads carry only the - * metric labels and values, so the collector stays a dumb recorder and the - * cursor code stays decoupled from `prom-client`. + * the durable scheduler's poll handler, consumed by the Prometheus poll-trigger + * collector. Payloads carry only the metric labels and values, so the collector + * stays a dumb recorder and the emitters stay decoupled from `prom-client`. */ /** Which cursor write was attempted. */ @@ -20,4 +20,8 @@ export type PollTriggerMetricsEventMap = { result: PollCursorCommitResult; durationMs: number; }; + + 'poll-tick-timed-out': { + nodeType: string; + }; }; diff --git a/packages/cli/src/metrics/prometheus/__tests__/poll-trigger-metrics.service.test.ts b/packages/cli/src/metrics/prometheus/__tests__/poll-trigger-metrics.service.test.ts index 0441a9e6d18..66ef86581bf 100644 --- a/packages/cli/src/metrics/prometheus/__tests__/poll-trigger-metrics.service.test.ts +++ b/packages/cli/src/metrics/prometheus/__tests__/poll-trigger-metrics.service.test.ts @@ -109,6 +109,7 @@ describe('PrometheusPollTriggerMetricsService', () => { expect.arrayContaining([ 'n8n_poll_trigger_errors_total', 'n8n_poll_trigger_overlapping_ticks_total', + 'n8n_poll_trigger_timeouts_total', 'n8n_poll_trigger_cursor_commits_total', ]), ); @@ -138,6 +139,7 @@ describe('PrometheusPollTriggerMetricsService', () => { 'poll-cursor-commit-settled', expect.any(Function), ); + expect(eventService.on).toHaveBeenCalledWith('poll-tick-timed-out', expect.any(Function)); }); }); @@ -197,6 +199,22 @@ describe('PrometheusPollTriggerMetricsService', () => { }); }); + describe('poll-tick-timed-out handler', () => { + it('counts the timed-out poll by node type', () => { + service.init(); + + const calls = eventService.on.mock.calls as unknown as Array< + [string, (payload: unknown) => void] + >; + const handler = calls.find((c) => c[0] === 'poll-tick-timed-out')![1]; + handler({ nodeType: 'n8n-nodes-base.testPoll' }); + + expect(counterIncFor('n8n_poll_trigger_timeouts_total')).toHaveBeenCalledWith({ + node_type: 'n8n-nodes-base.testPoll', + }); + }); + }); + describe('poll-cursor-commit-settled handler', () => { it('counts the commit by operation and result and observes its duration in seconds', () => { service.init(); diff --git a/packages/cli/src/metrics/prometheus/poll-trigger-metrics.service.ts b/packages/cli/src/metrics/prometheus/poll-trigger-metrics.service.ts index 42d726861ff..9dc2ab2a217 100644 --- a/packages/cli/src/metrics/prometheus/poll-trigger-metrics.service.ts +++ b/packages/cli/src/metrics/prometheus/poll-trigger-metrics.service.ts @@ -13,9 +13,9 @@ import { DURATION_BUCKETS_SECONDS } from './constant'; * for the poll reliability work. Opt-in via `includePollTriggerMetrics` and only * active on a main instance. Tick duration, errors, and same-process overlap come * from the core poll engine's event stream ({@link TriggersAndPollers.events}); - * cursor-commit outcomes come from `EventService`. Cross-instance overlap is not - * observable from inside a poll, so it is covered by the scheduler collector's - * `scheduler_tasks_lease_lost_total` instead. + * cursor-commit outcomes and scheduler-side poll timeouts come from `EventService`. + * Cross-instance overlap is not observable from inside a poll, so it is covered by + * the scheduler collector's `scheduler_tasks_lease_lost_total` instead. * * Labels are bounded (node type, status, kind, operation, result): no * workflow or instance label, per the metrics cardinality rule. @@ -55,6 +55,12 @@ export class PrometheusPollTriggerMetricsService implements PrometheusMetricsCol labelNames: ['node_type'], }); + const timeouts = new promClient.Counter({ + name: `${prefix}poll_trigger_timeouts_total`, + help: 'Total number of polls the durable scheduler abandoned after they exceeded N8N_SCHEDULER_POLL_TIMEOUT, by node type.', + labelNames: ['node_type'], + }); + const cursorCommits = new promClient.Counter({ name: `${prefix}poll_trigger_cursor_commits_total`, help: 'Total number of poll cursor commits by operation and result (success, fence_rejected, failure).', @@ -85,5 +91,9 @@ export class PrometheusPollTriggerMetricsService implements PrometheusMetricsCol cursorCommits.inc({ operation, result }); cursorCommitDuration.observe({ operation, result }, durationMs / 1000); }); + + this.eventService.on('poll-tick-timed-out', ({ nodeType }) => { + timeouts.inc({ node_type: nodeType }); + }); } } diff --git a/packages/cli/src/scheduling/__tests__/durable-scheduler.test.ts b/packages/cli/src/scheduling/__tests__/durable-scheduler.test.ts index cd092ab59aa..f77f1ff416f 100644 --- a/packages/cli/src/scheduling/__tests__/durable-scheduler.test.ts +++ b/packages/cli/src/scheduling/__tests__/durable-scheduler.test.ts @@ -31,6 +31,10 @@ describe('DurableScheduler', () => { executorIntervalSeconds = 5, materializationWindowSeconds = 60, misfireGraceSeconds = 60, + enabledForPollTriggers = false, + pollTimeoutSeconds = 45, + leaseDurationSeconds = 60, + useWorkflowPublicationService = true, } = {}) { const inner = mock(); vi.mocked(createScheduler).mockReturnValue(inner); @@ -61,7 +65,11 @@ describe('DurableScheduler', () => { minIntervalSeconds, materializationWindowSeconds, misfireGraceSeconds, + enabledForPollTriggers, + pollTimeoutSeconds, + leaseDurationSeconds, }, + workflows: { useWorkflowPublicationService }, }), tracing, scheduleTriggerTaskHandler, @@ -165,6 +173,78 @@ describe('DurableScheduler', () => { }); }); + describe('poll timeout warning', () => { + it('warns when a poll may outlive the lease on its occurrence', () => { + const { logger } = makeScheduler({ + enabledForPollTriggers: true, + pollTimeoutSeconds: 120, + leaseDurationSeconds: 60, + }); + + expect(logger.warn).toHaveBeenCalledWith( + expect.stringContaining('poll timeout'), + expect.objectContaining({ pollTimeoutSeconds: 120, leaseDurationSeconds: 60 }), + ); + }); + + // The poll deadline starts after the occurrence's setup reads, so a timeout + // equal to the lease already lets a full-length poll outlive it. + it('warns when the timeout equals the lease', () => { + const { logger } = makeScheduler({ + enabledForPollTriggers: true, + pollTimeoutSeconds: 60, + leaseDurationSeconds: 60, + }); + + expect(logger.warn).toHaveBeenCalledWith( + expect.stringContaining('poll timeout'), + expect.objectContaining({ pollTimeoutSeconds: 60, leaseDurationSeconds: 60 }), + ); + }); + + it('does not warn when the timeout fits inside the lease', () => { + const { logger } = makeScheduler({ + enabledForPollTriggers: true, + pollTimeoutSeconds: 45, + leaseDurationSeconds: 60, + }); + + expect(logger.warn).not.toHaveBeenCalledWith( + expect.stringContaining('poll timeout'), + expect.anything(), + ); + }); + + it('does not warn when poll triggers do not use the durable scheduler', () => { + const { logger } = makeScheduler({ + enabledForPollTriggers: false, + pollTimeoutSeconds: 120, + leaseDurationSeconds: 60, + }); + + expect(logger.warn).not.toHaveBeenCalledWith( + expect.stringContaining('poll timeout'), + expect.anything(), + ); + }); + + // Without the publication service the durable poller chain is inactive and + // polls run on the legacy in-memory path, where the timeout does not apply. + it('does not warn when the workflow publication service is disabled', () => { + const { logger } = makeScheduler({ + enabledForPollTriggers: true, + pollTimeoutSeconds: 120, + leaseDurationSeconds: 60, + useWorkflowPublicationService: false, + }); + + expect(logger.warn).not.toHaveBeenCalledWith( + expect.stringContaining('poll timeout'), + expect.anything(), + ); + }); + }); + describe('tracer', () => { // A fire span is opened from inside a timer callback armed while the claim // span was active, so it needs a fresh trace instead of parenting under a diff --git a/packages/cli/src/scheduling/durable-scheduler.ts b/packages/cli/src/scheduling/durable-scheduler.ts index a25f622f981..39218d0a879 100644 --- a/packages/cli/src/scheduling/durable-scheduler.ts +++ b/packages/cli/src/scheduling/durable-scheduler.ts @@ -14,6 +14,7 @@ import { InstanceSettings, Tracing } from 'n8n-core'; import { PrometheusSchedulerMetricsService } from '@/metrics/prometheus/scheduler-metrics.service'; import { withOwnerKeys } from './owner-key'; +import { isDurablePollerChainEnabled } from './poll-trigger-node/durable-poller-chain'; import { PollTriggerTaskHandler } from './poll-trigger-node/poll-trigger-task-handler'; import { ScheduleTriggerTaskHandler } from './schedule-trigger-node/schedule-trigger-task-handler'; import { createSchedulerTracer } from './scheduler-tracer'; @@ -92,6 +93,7 @@ export class DurableScheduler implements Scheduler { if (enabled) { warnOnMisfireGrace(logger, config); warnOnDrainRate(logger, config); + warnOnPollTimeout(logger, globalConfig); } this.registerTaskHandler(scheduleTriggerTaskHandler.taskType, scheduleTriggerTaskHandler); this.registerTaskHandler(pollTriggerTaskHandler.taskType, pollTriggerTaskHandler); @@ -162,6 +164,25 @@ function warnOnDrainRate(logger: Logger, config: GlobalConfig['scheduler']): voi } } +/** + * Warn when a poll may still be in flight after the lease on its occurrence has + * expired: the reaper can then reclaim the occurrence and another instance can + * start the same poll while the first one is still running. Equality counts + * too, since the poll deadline only starts after the occurrence's setup reads. + */ +function warnOnPollTimeout(logger: Logger, globalConfig: GlobalConfig): void { + const { pollTimeoutSeconds, leaseDurationSeconds } = globalConfig.scheduler; + if ( + isDurablePollerChainEnabled(globalConfig.scheduler, globalConfig.workflows) && + pollTimeoutSeconds >= leaseDurationSeconds + ) { + logger.warn( + 'Scheduler poll timeout reaches the lease duration; a poll can still be running when its lease expires and another instance takes the run over', + { pollTimeoutSeconds, leaseDurationSeconds }, + ); + } +} + export function buildMaterializerTransaction( dataSource: DataSource, jobs: ScheduledJobRepository, 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 988ccf12e88..1d923fba253 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,5 +1,6 @@ /* eslint-disable @typescript-eslint/unbound-method */ import type { Logger } from '@n8n/backend-common'; +import type { GlobalConfig } from '@n8n/config'; import type { PollerFailureState, WorkflowRepository } from '@n8n/db'; import { createDispatchReporter, type ClaimedTask } from '@n8n/scheduler'; import type { ErrorReporter, TriggersAndPollers } from 'n8n-core'; @@ -9,6 +10,7 @@ import { AsyncLocalStorage } from 'node:async_hooks'; import type { Mock, MockInstance } from 'vitest'; import { mock } from 'vitest-mock-extended'; +import type { EventService } from '@/events/event.service'; import { createNodeTypes } from '@/workflows/triggers/__tests__/trigger-test-utils'; import type { PollBackoffService } from '@/workflows/triggers/poll-backoff.service'; import type { TriggerExecutionContextFactory } from '@/workflows/triggers/trigger-execution-context.factory'; @@ -27,6 +29,10 @@ describe('PollTriggerTaskHandler', () => { const scopedLogger = mock(); const rootLogger = mock({ scoped: vi.fn().mockReturnValue(scopedLogger) }); + const eventService = mock(); + const pollTimeoutSeconds = 60; + const globalConfig = mock({ scheduler: { pollTimeoutSeconds } }); + const handler = new PollTriggerTaskHandler( rootLogger, triggerExecutionContextFactory, @@ -34,6 +40,8 @@ describe('PollTriggerTaskHandler', () => { workflowRepository, errorReporter, pollBackoffService, + eventService, + globalConfig, ); const onDispatch = vi.fn(); @@ -358,6 +366,118 @@ describe('PollTriggerTaskHandler', () => { }); }); + describe('poll timeout', () => { + const pollTimeoutMs = pollTimeoutSeconds * 1000; + + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + test('abandons a poll that outlives the timeout and reports no dispatch', async () => { + triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {})); + + const executing = handler.execute(buildTask(), report); + 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(); + expect(pollFunctions.__emitError).not.toHaveBeenCalled(); + expect(onDispatch).not.toHaveBeenCalled(); + expect(releaseIsolate).toHaveBeenCalledTimes(1); + expect(eventService.emit).toHaveBeenCalledWith('poll-tick-timed-out', { + nodeType: triggerNode.type, + }); + // The timeout counts as a transient poll failure, so a source that keeps + // hanging backs off like any failing source. + expect(pollBackoffService.recordFailure).toHaveBeenCalledWith( + expect.objectContaining({ + workflowId: 'wf-1', + nodeId: 'node-1', + error: expect.objectContaining({ failure: { cause: 'temporarily-unavailable' } }), + }), + ); + expect(pollBackoffService.recordSuccess).not.toHaveBeenCalled(); + }); + + test('records no failure for a workflow deactivated during a timed-out poll', async () => { + workflowRepository.isActive.mockResolvedValue(false); + triggersAndPollers.runPollFunction.mockReturnValue(new Promise(() => {})); + + const executing = handler.execute(buildTask(), report); + await vi.advanceTimersByTimeAsync(pollTimeoutMs); + await executing; + + 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; + }), + ); + + const executing = handler.execute(buildTask(), report); + 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); + }); + + test('discards the data of an abandoned poll that resolves after the timeout', async () => { + let resolvePoll: (data: INodeExecutionData[][]) => void = () => {}; + triggersAndPollers.runPollFunction.mockReturnValue( + new Promise((resolve) => { + resolvePoll = resolve; + }), + ); + + const executing = handler.execute(buildTask(), report); + await vi.advanceTimersByTimeAsync(pollTimeoutMs); + await executing; + resolvePoll(pollData); + await vi.advanceTimersByTimeAsync(0); + + // The tick was already reported as 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 () => { + let rejectPoll: (error: Error) => void = () => {}; + triggersAndPollers.runPollFunction.mockReturnValue( + new Promise((_resolve, reject) => { + rejectPoll = reject; + }), + ); + + const executing = handler.execute(buildTask(), report); + await vi.advanceTimersByTimeAsync(pollTimeoutMs); + await executing; + rejectPoll(new Error('poll source unreachable')); + await vi.advanceTimersByTimeAsync(0); + + // The tick was already reported as abandoned, so the late failure is dropped + // rather than routed to the error workflow. + expect(pollFunctions.__emitError).not.toHaveBeenCalled(); + expect(onDispatch).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 9063b87ac69..fa9de887d09 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,4 +1,6 @@ import { Logger } from '@n8n/backend-common'; +import { GlobalConfig } from '@n8n/config'; +import { Time } from '@n8n/constants'; import { WorkflowRepository } from '@n8n/db'; import { Service } from '@n8n/di'; import type { ClaimedTask, DispatchDecision, DispatchReporter, TaskHandler } from '@n8n/scheduler'; @@ -9,9 +11,10 @@ import { runPollInStagingScope, TriggersAndPollers, } from 'n8n-core'; -import type { INode, IWorkflowBase } from 'n8n-workflow'; -import { UnexpectedError } from 'n8n-workflow'; +import type { Failure, INode, IWorkflowBase } from 'n8n-workflow'; +import { OperationalError, UnexpectedError } from 'n8n-workflow'; +import { EventService } from '@/events/event.service'; import { PollBackoffService } from '@/workflows/triggers/poll-backoff.service'; import { TriggerExecutionContextFactory } from '@/workflows/triggers/trigger-execution-context.factory'; @@ -21,6 +24,28 @@ import { type PollTriggerTaskPayload, } from './poll-trigger-task'; +/** Race sentinel: `poll()` can resolve to anything, so the deadline resolves to a symbol it cannot produce. */ +const TIMED_OUT = Symbol('poll timed out'); + +/** Stands in for the error a hanging poll never threw, so backoff classifies the timeout as transient. */ +class PollTimeoutError extends OperationalError { + readonly failure: Failure = { cause: 'temporarily-unavailable' }; + + constructor() { + super('Poll exceeded its timeout and was abandoned'); + } +} + +/** An unref'd, cancellable deadline that resolves to {@link TIMED_OUT} after `ms`. */ +function timeoutAfter(ms: number): { timedOut: Promise; cancel: () => void } { + let timer: ReturnType | undefined; + const timedOut = new Promise((resolve) => { + timer = setTimeout(() => resolve(TIMED_OUT), ms); + timer.unref(); + }); + return { timedOut, cancel: () => clearTimeout(timer) }; +} + /** * Runs a due poll occurrence's `poll()` once and dispatches only when it returns new data. * Carries no `deduplicationKey`: under the at-least-once scheduler contract an occurrence @@ -30,6 +55,8 @@ import { export class PollTriggerTaskHandler implements TaskHandler { readonly taskType = POLL_TRIGGER_TASK_TYPE; + private readonly pollTimeoutMs: number; + constructor( private logger: Logger, private readonly triggerExecutionContextFactory: TriggerExecutionContextFactory, @@ -37,8 +64,11 @@ export class PollTriggerTaskHandler implements TaskHandler { private readonly workflowRepository: WorkflowRepository, 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): Promise { @@ -92,18 +122,59 @@ export class PollTriggerTaskHandler implements TaskHandler { // be committed by this poll and never by a later occurrence. return await runPollInStagingScope(pollFunctions, async () => { // Scheduled polls run outside any activation isolate window, so acquire and - // release one per tick; the finally releases even when poll() throws. + // release one per tick; the finally releases even when poll() throws, and + // even while an abandoned poll is still running. A late expression + // evaluation then fails and is discarded with the rest of that poll, + // whereas holding the isolate for a poll that may never settle would pin a + // pooled bridge for good. await workflow.expression.acquireIsolate(); // Nothing past a returning poll is the source failing, so a hand-off or // database error after it must not back the node off. A setup error before // poll() does count: it repeats every tick just like a failing source. let polled = false; try { - const pollResponse = await this.triggersAndPollers.runPollFunction( - workflow, - node, - pollFunctions, - ); + // `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 = 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 { + const outcome = await Promise.race([poll, deadline.timedOut]); + 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, + 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 + }); + } + return report.notDispatched(); + } + pollResponse = outcome; + } finally { + deadline.cancel(); + } polled = true; await this.pollBackoffService.recordSuccess({ workflowId, nodeId, state });