feat(core): Add settings for the timeout of scheduled tasks (no-changelog) (#40512)

packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts

@@ -42,6 +42,7 @@ describe('SchedulerConfig', () => {
 			expect(scheduler.triggerNodeMode).toBe('legacy');
 			expect(scheduler.allowSkipDurableScheduler).toBe(false);
 			expect(scheduler.maxAttempts).toBe(5);
+			expect(scheduler.taskTimeoutSeconds).toBe(300);
 			expect(scheduler.enabledForPollTriggers).toBe(false);
 			expect(scheduler.pollTimeoutSeconds).toBe(45);
 			expect(scheduler.enabledForSystemTasks).toBe(false);
@@ -74,6 +75,7 @@ describe('SchedulerConfig', () => {
 			vi.stubEnv('N8N_SCHEDULER_RETENTION_TIMEOUT', '120');
 			vi.stubEnv('N8N_SCHEDULER_MAX_CONCURRENT_PASSES', '4');
 			vi.stubEnv('N8N_SCHEDULER_MAX_ATTEMPTS', '3');
+			vi.stubEnv('N8N_SCHEDULER_TASK_TIMEOUT_SECONDS', '600');
 			vi.stubEnv('N8N_SCHEDULER_POLL_TRIGGERS_ENABLED', 'true');
 			vi.stubEnv('N8N_SCHEDULER_POLL_TIMEOUT', '30');
 			vi.stubEnv('N8N_SCHEDULER_SYSTEM_TASKS_ENABLED', 'true');
@@ -96,6 +98,7 @@ describe('SchedulerConfig', () => {
 			expect(scheduler.retentionTimeoutSeconds).toBe(120);
 			expect(scheduler.maxConcurrentPasses).toBe(4);
 			expect(scheduler.maxAttempts).toBe(3);
+			expect(scheduler.taskTimeoutSeconds).toBe(600);
 			expect(scheduler.enabledForPollTriggers).toBe(true);
 			expect(scheduler.pollTimeoutSeconds).toBe(30);
 			expect(scheduler.enabledForSystemTasks).toBe(true);
@@ -118,6 +121,15 @@ describe('SchedulerConfig', () => {
 			expect(scheduler.pollTimeoutSeconds).toBe(45);
 		});
 
+		it('should fall back to the default task timeout when the value exceeds the timer limit', () => {
+			vi.spyOn(console, 'warn').mockImplementation(() => {});
+			vi.stubEnv('N8N_SCHEDULER_TASK_TIMEOUT_SECONDS', '2147484');
+
+			const { scheduler } = Container.get(GlobalConfig);
+
+			expect(scheduler.taskTimeoutSeconds).toBe(300);
+		});
+
 		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

@@ -1,4 +1,4 @@
-import { DEFAULT_MISFIRE_GRACE_SECONDS, Time } from '@n8n/constants';
+import { DEFAULT_MISFIRE_GRACE_SECONDS, MAX_TIMER_DELAY_SECONDS, Time } from '@n8n/constants';
 import { z } from 'zod';
 
 import { Config, Env } from '../decorators';
@@ -276,6 +276,8 @@ export class SchedulerConfig {
 	 * 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.
+	 * A new value applies to a poll trigger the next time it is provisioned, for
+	 * example at startup or when its workflow is activated.
 	 * Must be greater than 0 and at most one day.
 	 */
 	@Env('N8N_SCHEDULER_POLL_TIMEOUT', positiveIntSchema.max(Time.days.toSeconds))
@@ -362,6 +364,20 @@ export class SchedulerConfig {
 	@Env('N8N_SCHEDULER_MAX_ATTEMPTS', positiveIntSchema)
 	maxAttempts: number = 5;
 
+	/**
+	 * How long, in seconds, one run of a scheduled task may take before the
+	 * scheduler aborts its signal and gives the run back. A handler that ignores
+	 * the signal keeps running. Defaults to 300 seconds.
+	 *
+	 * A timed-out run that had not started its work counts as a failed attempt (see
+	 * {@link maxAttempts}). Some tasks set their own timeout, which wins over this one.
+	 * A new value applies to a schedule the next time it is provisioned, for
+	 * example at startup or when its workflow is activated.
+	 * Must be greater than 0 and at most 2147483 (about 24 days).
+	 */
+	@Env('N8N_SCHEDULER_TASK_TIMEOUT_SECONDS', positiveIntSchema.max(MAX_TIMER_DELAY_SECONDS))
+	taskTimeoutSeconds: number = 5 * Time.minutes.toSeconds;
+
 	/**
 	 * How late, in seconds, a scheduled run may start and still count as on time. A
 	 * run later than this counts as missed, and the schedule's misfire policy decides

packages/@n8n/config/test/config.test.ts

@@ -509,6 +509,7 @@ describe('GlobalConfig', () => {
 			pollTimeoutSeconds: 45,
 			allowSkipDurableScheduler: false,
 			maxAttempts: 5,
+			taskTimeoutSeconds: 300,
 			misfireGraceSeconds: 60,
 			durableCursorsEnabled: false,
 			enabledForSystemTasks: false,

packages/@n8n/constants/src/scheduler.ts

@@ -121,6 +121,9 @@ export type ScheduledJobMisfirePolicy =
  */
 export const DEFAULT_MISFIRE_GRACE_SECONDS = 60;
 
+/** Longest delay, in seconds, a Node.js timer honors. Node fires a longer one after about 1 ms. */
+export const MAX_TIMER_DELAY_SECONDS = Math.floor((2 ** 31 - 1) / 1000);
+
 /**
  * Where a scheduled task is in its lifecycle, from waiting to run to a final outcome.
  */

packages/@n8n/decorators/src/system-task/__tests__/run-options.test.ts

@@ -1,7 +1,8 @@
-import { ScheduledJobMisfirePolicy } from '@n8n/constants';
+import { MAX_TIMER_DELAY_SECONDS, ScheduledJobMisfirePolicy } from '@n8n/constants';
 
 import {
 	resolveSystemTaskRunOptions,
+	timeoutAfterLimit,
 	validateSystemTask,
 	type SystemTask,
 	type SystemTaskEffects,
@@ -44,6 +45,7 @@ it.each([
 	['misfirePolicy', { misfirePolicy: ScheduledJobMisfirePolicy.Skip }],
 	['misfireGraceSeconds', { misfireGraceSeconds: 5 }],
 	['concurrencyLimit', { concurrencyLimit: 4 }],
+	['timeoutSeconds', { timeoutSeconds: 600 }],
 ] as const)('should let a task override %s', (field, override) => {
 	const options = resolveSystemTaskRunOptions(taskWith({ effects: 'idempotent', ...override }));
 
@@ -63,6 +65,10 @@ it.each([
 	{ concurrencyLimit: -1 },
 	{ concurrencyLimit: 1.5 },
 	{ concurrencyLimit: 2_147_483_648 },
+	{ timeoutSeconds: 0 },
+	{ timeoutSeconds: -1 },
+	{ timeoutSeconds: 1.5 },
+	{ timeoutSeconds: 2_147_484 },
 ])('should reject the nonsensical override %o', (override) => {
 	expect(() =>
 		resolveSystemTaskRunOptions(taskWith({ effects: 'idempotent', ...override })),
@@ -95,6 +101,20 @@ it('should refuse to retry non-idempotent work that asked for more attempts', ()
 	expect(options.maxAttempts).toBe(1);
 });
 
+it('should leave the timeout to the scheduler when a task does not override it', () => {
+	const options = resolveSystemTaskRunOptions(taskWith({ effects: 'idempotent' }));
+
+	expect(options.timeoutSeconds).toBeUndefined();
+});
+
+it('should accept the longest timeout a timer honors', () => {
+	const options = resolveSystemTaskRunOptions(
+		taskWith({ effects: 'idempotent', timeoutSeconds: 2_147_483 }),
+	);
+
+	expect(options.timeoutSeconds).toBe(2_147_483);
+});
+
 it('should let a task permit overlap', () => {
 	const options = resolveSystemTaskRunOptions(
 		taskWith({ effects: 'idempotent', concurrencyLimit: null }),
@@ -150,3 +170,17 @@ it('should accept a cluster task interval of 0 seconds, which is rounded up', ()
 		),
 	).not.toThrow();
 });
+
+describe('timeoutAfterLimit', () => {
+	it.each([
+		{ limit: 300, margin: 300, expected: 600 },
+		{ limit: 1.2, margin: 0, expected: 2 },
+		{ limit: MAX_TIMER_DELAY_SECONDS, margin: 300, expected: MAX_TIMER_DELAY_SECONDS },
+	])('adds the margin to a limit of $limit', ({ limit, margin, expected }) => {
+		expect(timeoutAfterLimit(limit, margin)).toBe(expected);
+	});
+
+	it.each([0, -1, Number.NaN])('uses the longest timeout for no limit (%s)', (limit) => {
+		expect(timeoutAfterLimit(limit, 300)).toBe(MAX_TIMER_DELAY_SECONDS);
+	});
+});

packages/@n8n/decorators/src/system-task/index.ts

@@ -5,6 +5,7 @@ export {
 	intervalFromSeconds,
 	resolveSystemTaskRunOptions,
 	resolveSystemTaskSchedule,
+	timeoutAfterLimit,
 	validateSystemTask,
 } from './system-task';
 export type {

packages/@n8n/decorators/src/system-task/system-task.ts

@@ -1,6 +1,7 @@
 import {
 	DEFAULT_MISFIRE_GRACE_SECONDS,
 	MAX_INTEGER_32BITS_SIGNED,
+	MAX_TIMER_DELAY_SECONDS,
 	ScheduledJobMisfirePolicy,
 	Time,
 	type IntervalDefinition,
@@ -74,9 +75,16 @@ export interface SystemTask {
 	readonly concurrencyLimit?: number | null;
 
 	/**
-	 * Executes one occurrence of the task. A run may take as long as it needs.
+	 * Overrides how long, in seconds, one durable run may take before the
+	 * scheduler aborts its signal and gives the occurrence back. An integer from 1
+	 * to {@link MAX_TIMER_DELAY_SECONDS}. Defaults to N8N_SCHEDULER_TASK_TIMEOUT_SECONDS.
+	 */
+	readonly timeoutSeconds?: number;
+
+	/**
+	 * Executes one occurrence of the task. A durable run may take up to its timeout.
 	 * `signal` aborts on shutdown, and on loss of leadership for an in-memory timer.
-	 * A durable run also aborts on lease loss or expiry.
+	 * A durable run also aborts on lease loss or expiry, and when it reaches its timeout.
 	 * Ignoring the signal delays shutdown and can let another run overlap.
 	 * A durable run that settles after a lease abort counts as a failed attempt
 	 * while its claim still matches, and retries only while attempts remain.
@@ -95,6 +103,8 @@ export interface SystemTaskRunOptions {
 	maxAttempts: number;
 	/** `null` means no limit. */
 	concurrencyLimit: number | null;
+	/** `undefined` means the scheduler's task timeout. */
+	timeoutSeconds: number | undefined;
 }
 
 /** One occurrence at a time, like the in-memory timer. */
@@ -107,7 +117,7 @@ export const DEFAULT_SYSTEM_TASK_CONCURRENCY_LIMIT = 1;
  */
 const SYSTEM_TASK_RUN_OPTION_DEFAULTS: Record<
 	SystemTaskEffects,
-	Omit<SystemTaskRunOptions, 'misfireGraceSeconds' | 'concurrencyLimit'>
+	Omit<SystemTaskRunOptions, 'misfireGraceSeconds' | 'concurrencyLimit' | 'timeoutSeconds'>
 > = {
 	idempotent: {
 		misfirePolicy: ScheduledJobMisfirePolicy.Coalesce,
@@ -137,6 +147,7 @@ export function resolveSystemTaskRunOptions(task: SystemTask): SystemTaskRunOpti
 			task.concurrencyLimit === undefined
 				? DEFAULT_SYSTEM_TASK_CONCURRENCY_LIMIT
 				: task.concurrencyLimit,
+		timeoutSeconds: task.timeoutSeconds,
 	};
 
 	// These end up in `int` columns, where a fractional value is rounded and anything
@@ -151,17 +162,17 @@ export function resolveSystemTaskRunOptions(task: SystemTask): SystemTaskRunOpti
 	if (options.concurrencyLimit !== null) {
 		assertInRange(task.name, 'concurrencyLimit', options.concurrencyLimit, 1);
 	}
+	if (options.timeoutSeconds !== undefined) {
+		assertInRange(task.name, 'timeoutSeconds', options.timeoutSeconds, 1, MAX_TIMER_DELAY_SECONDS);
+	}
 
 	return options;
 }
 
-/** Longest delay a timeout honors. Node fires a longer one after about 1 ms. */
-const MAX_RETRY_DELAY_SECONDS = Math.floor(MAX_INTEGER_32BITS_SIGNED / Time.seconds.toMilliseconds);
-
 /**
  * Rejects a task that declares an option the schedulers cannot honor.
  *
- * @throws {UnexpectedError} when `retryDelaySeconds`, `maxAttempts` or `misfireGraceSeconds` is out of range
+ * @throws {UnexpectedError} when `retryDelaySeconds`, `maxAttempts`, `misfireGraceSeconds` or `timeoutSeconds` is out of range
  * @throws {UnexpectedError} when an instance task declares an interval that is not positive and finite
  */
 export function validateSystemTask(task: SystemTask): void {
@@ -172,7 +183,7 @@ export function validateSystemTask(task: SystemTask): void {
 		retryDelaySeconds !== undefined &&
 		(!Number.isInteger(retryDelaySeconds) ||
 			retryDelaySeconds < 1 ||
-			retryDelaySeconds > MAX_RETRY_DELAY_SECONDS)
+			retryDelaySeconds > MAX_TIMER_DELAY_SECONDS)
 	) {
 		throw new UnexpectedError('A system task declares an out-of-range retry delay', {
 			extra: { name: task.name, retryDelaySeconds },
@@ -210,6 +221,18 @@ export function intervalFromSeconds(seconds: number): IntervalDefinition {
 	return { kind: 'interval', intervalSeconds: Math.round(seconds) };
 }
 
+/**
+ * The timeout of a durable run that another limit already stops: that limit plus
+ * `marginSeconds`, rounded up to the whole second. A limit of 0 or less means the
+ * run has no limit, so the timeout is the longest the scheduler enforces.
+ */
+export function timeoutAfterLimit(limitSeconds: number, marginSeconds: number): number {
+	if (!(limitSeconds > 0)) {
+		return MAX_TIMER_DELAY_SECONDS;
+	}
+	return Math.min(Math.ceil(limitSeconds + marginSeconds), MAX_TIMER_DELAY_SECONDS);
+}
+
 /** An interval schedule firing every `milliseconds`, rounded to the whole millisecond. */
 export function intervalFromMilliseconds(milliseconds: number): IntervalDefinition {
 	return {
@@ -248,10 +271,16 @@ function wholeMilliseconds(seconds: number): number {
 	);
 }
 
-function assertInRange(taskName: string, field: string, value: number, min: number) {
-	if (!Number.isInteger(value) || value < min || value > MAX_INTEGER_32BITS_SIGNED) {
+function assertInRange(
+	taskName: string,
+	field: string,
+	value: number,
+	min: number,
+	max = MAX_INTEGER_32BITS_SIGNED,
+) {
+	if (!Number.isInteger(value) || value < min || value > max) {
 		throw new UnexpectedError('A system task declares an out-of-range option', {
-			extra: { name: taskName, field, value, min, max: MAX_INTEGER_32BITS_SIGNED },
+			extra: { name: taskName, field, value, min, max },
 		});
 	}
 }