feat(core): Bound a durable-scheduler poll tick with a configurable timeout (no-changelog) (#35874)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Lorent Lempereur <lorent.lempereur@n8n.io>

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');
 

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

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,

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;
+	};
 };

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();

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 });
+		});
 	}
 }

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<Scheduler & SchedulerPasses>();
 		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

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,

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<Logger>();
 	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,
 		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' } });

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<typeof TIMED_OUT>; cancel: () => void } {
+	let timer: ReturnType<typeof setTimeout> | undefined;
+	const timedOut = new Promise<typeof TIMED_OUT>((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<DispatchDecision> {
@@ -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<typeof poll>;
+				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 });