From 2a1fde3634d422bde0176af3c8a4f640869d4e77 Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Mon, 12 Oct 2026 10:46:09 +0000 Subject: [PATCH] feat(core): Add settings for the timeout of scheduled tasks (no-changelog) (#40512) --- .../__tests__/scheduler.config.test.ts | 12 +++++ .../config/src/configs/scheduler.config.ts | 18 ++++++- packages/@n8n/config/test/config.test.ts | 1 + packages/@n8n/constants/src/scheduler.ts | 3 ++ .../system-task/__tests__/run-options.test.ts | 36 ++++++++++++- .../@n8n/decorators/src/system-task/index.ts | 1 + .../decorators/src/system-task/system-task.ts | 51 +++++++++++++++---- 7 files changed, 109 insertions(+), 13 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 cfb72a3c1e3..6fd48056447 100644 --- a/packages/@n8n/config/src/configs/__tests__/scheduler.config.test.ts +++ b/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'); diff --git a/packages/@n8n/config/src/configs/scheduler.config.ts b/packages/@n8n/config/src/configs/scheduler.config.ts index 2b0ffc60c2a..517ec5ba009 100644 --- a/packages/@n8n/config/src/configs/scheduler.config.ts +++ b/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 diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index b280ce2a61c..d32aadbb7e0 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/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, diff --git a/packages/@n8n/constants/src/scheduler.ts b/packages/@n8n/constants/src/scheduler.ts index 4f0b4931df1..406be86dfd1 100644 --- a/packages/@n8n/constants/src/scheduler.ts +++ b/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. */ diff --git a/packages/@n8n/decorators/src/system-task/__tests__/run-options.test.ts b/packages/@n8n/decorators/src/system-task/__tests__/run-options.test.ts index 670ef50aa97..70e2fe2bb51 100644 --- a/packages/@n8n/decorators/src/system-task/__tests__/run-options.test.ts +++ b/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); + }); +}); diff --git a/packages/@n8n/decorators/src/system-task/index.ts b/packages/@n8n/decorators/src/system-task/index.ts index ff0c42795ed..a12afece732 100644 --- a/packages/@n8n/decorators/src/system-task/index.ts +++ b/packages/@n8n/decorators/src/system-task/index.ts @@ -5,6 +5,7 @@ export { intervalFromSeconds, resolveSystemTaskRunOptions, resolveSystemTaskSchedule, + timeoutAfterLimit, validateSystemTask, } from './system-task'; export type { diff --git a/packages/@n8n/decorators/src/system-task/system-task.ts b/packages/@n8n/decorators/src/system-task/system-task.ts index 602e4efce80..6a79f5f6bf2 100644 --- a/packages/@n8n/decorators/src/system-task/system-task.ts +++ b/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 + Omit > = { 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 }, }); } }