From 6e12447f9b747da8f41cf7246e57ca0fa7ed5f16 Mon Sep 17 00:00:00 2001 From: "n8n-assistant[bot]" <100856346+n8n-assistant[bot]@users.noreply.github.com> Date: Thu, 3 Sep 2026 15:20:05 +0000 Subject: [PATCH] fix(core): Cap task runner task timeouts to the graceful shutdown window (backport to release-candidate/2.38.x) (#37765) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Mike Repeć Co-authored-by: Claude Fable 5 --- .../__tests__/task-runner-process.test.ts | 40 ++++- .../__tests__/task-broker-ws-server.test.ts | 44 +++++- .../__tests__/task-broker.service.test.ts | 140 ++++++++++++++++++ .../task-runner-shutdown-timeout.error.ts | 10 ++ .../task-broker/task-broker-ws-server.ts | 29 +++- .../task-broker/task-broker.service.ts | 71 +++++++-- .../task-runners/task-runner-process-base.ts | 11 +- 7 files changed, 329 insertions(+), 16 deletions(-) create mode 100644 packages/cli/src/task-runners/task-broker/errors/task-runner-shutdown-timeout.error.ts diff --git a/packages/cli/src/task-runners/__tests__/task-runner-process.test.ts b/packages/cli/src/task-runners/__tests__/task-runner-process.test.ts index 82409af3dae..5558fbb01a6 100644 --- a/packages/cli/src/task-runners/__tests__/task-runner-process.test.ts +++ b/packages/cli/src/task-runners/__tests__/task-runner-process.test.ts @@ -9,7 +9,7 @@ import type { TaskBrokerAuthService } from '@/task-runners/task-broker/auth/task import { JsTaskRunnerProcess } from '@/task-runners/task-runner-process-js'; import { TaskRunnerLifecycleEvents } from '../task-runner-lifecycle-events'; -import { restartRetryDelay } from '../task-runner-process-base'; +import { restartRetryDelay, RUNNER_EXIT_GRACE_MS } from '../task-runner-process-base'; // Source imports `spawn` from `node:child_process` as an ESM binding, so mutating // `require('child_process').spawn` does not intercept it — mock the module instead. @@ -98,6 +98,44 @@ describe('TaskRunnerProcess', () => { }); }); + describe('stop', () => { + let child: ChildProcess; + + beforeEach(async () => { + vi.useFakeTimers(); + taskRunnerProcess = new JsTaskRunnerProcess(logger, runnerConfig, authService, mock()); + authService.createGrantToken.mockResolvedValue('grantToken'); + child = createChildProcess(42); + spawnMock.mockReturnValue(child); + await taskRunnerProcess.start(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('should not force-kill a runner that exits within the grace period', async () => { + const stopPromise = taskRunnerProcess.stop(); + await vi.advanceTimersByTimeAsync(0); // let stop() send the graceful kill + child.emit('exit', 0); + await stopPromise; + await vi.advanceTimersByTimeAsync(RUNNER_EXIT_GRACE_MS); // no late force-kill + + expect(child.kill).toHaveBeenCalledTimes(1); + expect(child.kill).not.toHaveBeenCalledWith('SIGKILL'); + }); + + it('should force-kill a runner that does not exit within the grace period', async () => { + const stopPromise = taskRunnerProcess.stop(); + await vi.advanceTimersByTimeAsync(RUNNER_EXIT_GRACE_MS); + + expect(child.kill).toHaveBeenCalledWith('SIGKILL'); + + child.emit('exit', 0); + await stopPromise; + }); + }); + describe('start', () => { beforeEach(() => { taskRunnerProcess = new JsTaskRunnerProcess(logger, runnerConfig, authService, mock()); diff --git a/packages/cli/src/task-runners/task-broker/__tests__/task-broker-ws-server.test.ts b/packages/cli/src/task-runners/task-broker/__tests__/task-broker-ws-server.test.ts index 9fa3ba69006..544219057f8 100644 --- a/packages/cli/src/task-runners/task-broker/__tests__/task-broker-ws-server.test.ts +++ b/packages/cli/src/task-runners/task-broker/__tests__/task-broker-ws-server.test.ts @@ -4,10 +4,16 @@ import { Time } from '@n8n/constants'; import { mock } from 'vitest-mock-extended'; import type WebSocket from 'ws'; -import { WsStatusCodes } from '@/constants'; +import { ShutdownMetadata } from '@n8n/decorators'; +import { Container } from '@n8n/di'; + +import { HIGHEST_SHUTDOWN_PRIORITY, WsStatusCodes } from '@/constants'; import type { EventService } from '@/events/event.service'; import type { DefaultTaskRunnerDisconnectAnalyzer } from '@/task-runners/default-task-runner-disconnect-analyzer'; -import { TaskBrokerWsServer } from '@/task-runners/task-broker/task-broker-ws-server'; +import { + SHUTDOWN_TASK_BUDGET_RATIO, + TaskBrokerWsServer, +} from '@/task-runners/task-broker/task-broker-ws-server'; import type { TaskBroker } from '@/task-runners/task-broker/task-broker.service'; import { TaskRunnerLifecycleEvents } from '@/task-runners/task-runner-lifecycle-events'; @@ -276,6 +282,40 @@ describe('TaskBrokerWsServer', () => { }); }); + describe('capTaskTimeoutsForShutdown', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('should register at the highest shutdown priority, so the cap runs alongside the worker drain instead of after it', () => { + const handlers = + Container.get(ShutdownMetadata).getHandlersByPriority()[HIGHEST_SHUTDOWN_PRIORITY] ?? []; + + expect(handlers).toContainEqual({ + serviceClass: TaskBrokerWsServer, + methodName: 'capTaskTimeoutsForShutdown', + }); + }); + + it('should cap broker task timeouts to 80% of the graceful shutdown window', () => { + const taskBroker = mock(); + const server = createServer({ taskBroker }); + + server.capTaskTimeoutsForShutdown(); + + expect(taskBroker.capTaskTimeoutsForShutdown).toHaveBeenCalledWith( + Date.now() + + globalConfig.generic.gracefulShutdownTimeout * + SHUTDOWN_TASK_BUDGET_RATIO * + Time.seconds.toMilliseconds, + ); + }); + }); + describe('heartbeat timer', () => { const DEAD = false; diff --git a/packages/cli/src/task-runners/task-broker/__tests__/task-broker.service.test.ts b/packages/cli/src/task-runners/task-broker/__tests__/task-broker.service.test.ts index 4afa320f7a5..392cf7ab2c9 100644 --- a/packages/cli/src/task-runners/task-broker/__tests__/task-broker.service.test.ts +++ b/packages/cli/src/task-runners/task-broker/__tests__/task-broker.service.test.ts @@ -10,6 +10,7 @@ import type { TaskRunnerLifecycleEvents } from '@/task-runners/task-runner-lifec import { TaskRejectError } from '../errors/task-reject.error'; import { TaskRequesterAcceptTimeoutError } from '../errors/task-requester-accept-timeout.error'; import { TaskRunnerExecutionTimeoutError } from '../errors/task-runner-execution-timeout.error'; +import { TaskRunnerShutdownTimeoutError } from '../errors/task-runner-shutdown-timeout.error'; import { TaskRunnerUnreachableError } from '../errors/task-runner-unreachable.error'; import { TaskBroker } from '../task-broker.service'; import type { @@ -1248,6 +1249,145 @@ describe('TaskBroker', () => { }); }); + describe('capTaskTimeoutsForShutdown', () => { + let taskBroker: TaskBroker; + let config: TaskRunnersConfig; + let runnerLifecycleEvents: MockProxy; + + const taskId = 'task1'; + const runnerId = 'runner1'; + const requesterId = 'requester1'; + let requesterCallback: ReturnType>; + + const setupTaskWithArmedTimer = async () => { + taskBroker.registerRunner(mock({ id: runnerId }), vi.fn()); + taskBroker.registerRequester(requesterId, requesterCallback); + taskBroker.setTasks({ + [taskId]: { id: taskId, runnerId, requesterId, taskType: 'test' }, + }); + await taskBroker.sendTaskSettings(taskId, {}); // arms the 300s timer + }; + + beforeEach(() => { + vi.useFakeTimers(); + runnerLifecycleEvents = mock(); + requesterCallback = vi.fn(); + config = mock({ + taskTimeout: 300, + taskRequestTimeout: 60, + taskAcceptTimeout: 2, + mode: 'external', + }); + taskBroker = new TaskBroker(mock(), config, runnerLifecycleEvents, mock()); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('should re-arm a timer due after the deadline so the task fails at the deadline', async () => { + await setupTaskWithArmedTimer(); + + taskBroker.capTaskTimeoutsForShutdown(Date.now() + 96_000); + + await vi.advanceTimersByTimeAsync(96_000); + + expect(requesterCallback).toHaveBeenCalledWith({ + type: 'broker:taskerror', + taskId, + error: expect.any(TaskRunnerShutdownTimeoutError), + }); + expect(taskBroker.getTasks().get(taskId)).toBeUndefined(); + }); + + it('should never extend a timer already due before the deadline', async () => { + taskBroker.registerRunner(mock({ id: runnerId }), vi.fn()); + taskBroker.registerRequester(requesterId, requesterCallback); + const timesOutAt = Date.now() + 10_000; + taskBroker.setTasks({ + [taskId]: { + id: taskId, + runnerId, + requesterId, + taskType: 'test', + timesOutAt, + timeout: setTimeout(async () => { + await taskBroker.taskErrorHandler(taskId, new TaskRunnerExecutionTimeoutError(mock())); + }, 10_000), + }, + }); + + taskBroker.capTaskTimeoutsForShutdown(Date.now() + 96_000); + + expect(taskBroker.getTasks().get(taskId)?.timesOutAt).toBe(timesOutAt); + + await vi.advanceTimersByTimeAsync(10_000); + + expect(requesterCallback).toHaveBeenCalledWith({ + type: 'broker:taskerror', + taskId, + error: expect.any(TaskRunnerExecutionTimeoutError), + }); + }); + + it('should cap timers armed after the deadline was set', async () => { + taskBroker.capTaskTimeoutsForShutdown(Date.now() + 96_000); + + await setupTaskWithArmedTimer(); + + expect(taskBroker.getTasks().get(taskId)?.timesOutAt).toBe(Date.now() + 96_000); + + await vi.advanceTimersByTimeAsync(96_000); + + expect(requesterCallback).toHaveBeenCalledWith({ + type: 'broker:taskerror', + taskId, + error: expect.any(TaskRunnerShutdownTimeoutError), + }); + }); + + it('[internal mode] should not restart the runner when a capped timer fires', async () => { + config = mock({ + taskTimeout: 300, + taskRequestTimeout: 60, + taskAcceptTimeout: 2, + mode: 'internal', + }); + taskBroker = new TaskBroker(mock(), config, runnerLifecycleEvents, mock()); + await setupTaskWithArmedTimer(); + + taskBroker.capTaskTimeoutsForShutdown(Date.now() + 96_000); + + await vi.advanceTimersByTimeAsync(96_000); + + expect(runnerLifecycleEvents.emit).not.toHaveBeenCalledWith( + 'runner:timed-out-during-task', + expect.anything(), + ); + expect(requesterCallback).toHaveBeenCalledWith({ + type: 'broker:taskerror', + taskId, + error: expect.any(TaskRunnerShutdownTimeoutError), + }); + }); + + it('should not fail a task that completes before the deadline', async () => { + await setupTaskWithArmedTimer(); + + taskBroker.capTaskTimeoutsForShutdown(Date.now() + 96_000); + + await taskBroker.taskDoneHandler(taskId, { result: [] }); + await vi.advanceTimersByTimeAsync(96_000); + + expect(requesterCallback).toHaveBeenCalledTimes(1); + expect(requesterCallback).toHaveBeenCalledWith({ + type: 'broker:taskdone', + taskId, + data: { result: [] }, + }); + }); + }); + describe('task runner accept timeout', () => { const ACCEPT_TIMEOUT_MS = 2100; diff --git a/packages/cli/src/task-runners/task-broker/errors/task-runner-shutdown-timeout.error.ts b/packages/cli/src/task-runners/task-broker/errors/task-runner-shutdown-timeout.error.ts new file mode 100644 index 00000000000..0e0ac92fe73 --- /dev/null +++ b/packages/cli/src/task-runners/task-broker/errors/task-runner-shutdown-timeout.error.ts @@ -0,0 +1,10 @@ +import { OperationalError } from 'n8n-workflow'; + +export class TaskRunnerShutdownTimeoutError extends OperationalError { + description = + 'The n8n instance began shutting down while the task was still running, so n8n stopped the task to finish shutting down in time. Retry the execution, or catch this error in an error workflow.'; + + constructor() { + super('Task aborted because the n8n instance was shutting down'); + } +} diff --git a/packages/cli/src/task-runners/task-broker/task-broker-ws-server.ts b/packages/cli/src/task-runners/task-broker/task-broker-ws-server.ts index 3d49f25ac86..5fb28bb5f1b 100644 --- a/packages/cli/src/task-runners/task-broker/task-broker-ws-server.ts +++ b/packages/cli/src/task-runners/task-broker/task-broker-ws-server.ts @@ -1,13 +1,14 @@ import { Logger } from '@n8n/backend-common'; import { GlobalConfig, TaskRunnersConfig } from '@n8n/config'; import { Time } from '@n8n/constants'; +import { OnShutdown } from '@n8n/decorators'; import { Service } from '@n8n/di'; import type { BrokerMessage, RunnerMessage } from '@n8n/task-runner'; import { sleep } from '@n8n/utils/sleep'; import { jsonStringify, UserError } from 'n8n-workflow'; import type WebSocket from 'ws'; -import { WsStatusCodes } from '@/constants'; +import { HIGHEST_SHUTDOWN_PRIORITY, WsStatusCodes } from '@/constants'; import { EventService } from '@/events/event.service'; import { DefaultTaskRunnerDisconnectAnalyzer } from '@/task-runners/default-task-runner-disconnect-analyzer'; import type { @@ -27,6 +28,12 @@ function heartbeat(this: WebSocket) { this.isAlive = true; } +/** + * Share of the graceful-shutdown window that task work may use. The remainder is + * the budget for failed executions to persist before the force-kill timer fires. + */ +export const SHUTDOWN_TASK_BUDGET_RATIO = 0.8; + type WsStatusCode = (typeof WsStatusCodes)[keyof typeof WsStatusCodes]; type RemoveConnectionOptions = { @@ -104,6 +111,22 @@ export class TaskBrokerWsServer { } } + /** + * Runs at shutdown start, concurrently with the worker's execution drain. The + * regular `stop()` runs only after that drain - too late for a task timeout that + * outlasts the force-kill deadline while an execution stuck on it blocks the drain. + */ + @OnShutdown(HIGHEST_SHUTDOWN_PRIORITY) + capTaskTimeoutsForShutdown() { + const deadline = + Date.now() + + this.globalConfig.generic.gracefulShutdownTimeout * + SHUTDOWN_TASK_BUDGET_RATIO * + Time.seconds.toMilliseconds; + + this.taskBroker.capTaskTimeoutsForShutdown(deadline); + } + async stop() { this.runnerLifecycleEvents.off('runner:unresponsive', this.onRunnerUnresponsive); @@ -292,7 +315,9 @@ export class TaskBrokerWsServer { } private async drainActiveTasks() { - const drainTimeout = Math.floor(this.globalConfig.generic.gracefulShutdownTimeout * 0.8); + const drainTimeout = Math.floor( + this.globalConfig.generic.gracefulShutdownTimeout * SHUTDOWN_TASK_BUDGET_RATIO, + ); const drainTimeoutMs = drainTimeout * Time.seconds.toMilliseconds; diff --git a/packages/cli/src/task-runners/task-broker/task-broker.service.ts b/packages/cli/src/task-runners/task-broker/task-broker.service.ts index be0b8623029..c17d75ddcd2 100644 --- a/packages/cli/src/task-runners/task-broker/task-broker.service.ts +++ b/packages/cli/src/task-runners/task-broker/task-broker.service.ts @@ -16,6 +16,7 @@ import { TaskRejectError } from '@/task-runners/task-broker/errors/task-reject.e import { TaskRequesterAcceptTimeoutError } from '@/task-runners/task-broker/errors/task-requester-accept-timeout.error'; import { TaskRunnerAcceptTimeoutError } from '@/task-runners/task-broker/errors/task-runner-accept-timeout.error'; import { TaskRunnerExecutionTimeoutError } from '@/task-runners/task-broker/errors/task-runner-execution-timeout.error'; +import { TaskRunnerShutdownTimeoutError } from '@/task-runners/task-broker/errors/task-runner-shutdown-timeout.error'; import { TaskRunnerUnreachableError } from '@/task-runners/task-broker/errors/task-runner-unreachable.error'; import { TaskRunnerLifecycleEvents } from '@/task-runners/task-runner-lifecycle-events'; @@ -32,6 +33,8 @@ export interface Task { requesterId: string; taskType: string; timeout?: NodeJS.Timeout; + /** Epoch ms when `timeout` fires. Lets the shutdown cap skip timers already due sooner. */ + timesOutAt?: number; } export interface TaskOffer { @@ -98,6 +101,9 @@ export class TaskBroker { */ private isDraining = false; + /** Epoch ms by which every task timeout must fire once shutdown has begun. */ + private shutdownDeadline?: number; + private runnerAcceptRejects: Map< Task['id'], { accept: RunnerAcceptCallback; reject: TaskRejectCallback; runnerId: TaskRunner['id'] } @@ -594,9 +600,11 @@ export class TaskBroker { const task = this.tasks.get(taskId); if (!task) return; + const taskTimeoutMs = this.taskRunnersConfig.taskTimeout * Time.seconds.toMilliseconds; + task.timesOutAt = Math.min(Date.now() + taskTimeoutMs, this.shutdownDeadline ?? Infinity); task.timeout = setTimeout(async () => { await this.handleTaskTimeout(taskId); - }, this.taskRunnersConfig.taskTimeout * Time.seconds.toMilliseconds); + }, task.timesOutAt - Date.now()); await this.messageRunner(runner.id, { type: 'broker:tasksettings', @@ -609,15 +617,24 @@ export class TaskBroker { const task = this.tasks.get(taskId); if (!task) return; + // A capped timer fires at the shutdown deadline, not the task's own timeout. + const isCappedByShutdown = + this.shutdownDeadline !== undefined && task.timesOutAt === this.shutdownDeadline; + if (this.taskRunnersConfig.mode === 'internal') { - this.taskRunnerLifecycleEvents.emit('runner:timed-out-during-task', { - runnerId: task.runnerId, - }); + // Don't restart a runner that shutdown is about to stop anyway. + if (this.shutdownDeadline === undefined) { + this.taskRunnerLifecycleEvents.emit('runner:timed-out-during-task', { + runnerId: task.runnerId, + }); + } } else if (this.taskRunnersConfig.mode === 'external') { await this.messageRunner(task.runnerId, { type: 'broker:taskcancel', taskId, - reason: 'Task execution timed out', + reason: isCappedByShutdown + ? 'Task aborted because this n8n instance is shutting down' + : 'Task execution timed out', }); } @@ -625,11 +642,13 @@ export class TaskBroker { await this.taskErrorHandler( taskId, - new TaskRunnerExecutionTimeoutError({ - taskTimeout, - isSelfHosted: this.globalConfig.deployment.type !== 'cloud', - mode, - }), + isCappedByShutdown + ? new TaskRunnerShutdownTimeoutError() + : new TaskRunnerExecutionTimeoutError({ + taskTimeout, + isSelfHosted: this.globalConfig.deployment.type !== 'cloud', + mode, + }), ); } @@ -979,6 +998,38 @@ export class TaskBroker { return this.tasks.size > 0; } + /** + * Caps every task timeout, current and future, to fire no later than `deadline`, + * without extending timers already due sooner. A task that fails inside the + * shutdown window lets its execution error normally and the worker drain complete. + */ + capTaskTimeoutsForShutdown(deadline: number) { + this.shutdownDeadline = deadline; + + const cappedTaskIds: Array = []; + + for (const [taskId, task] of this.tasks) { + if (!task.timeout) continue; + if (task.timesOutAt !== undefined && task.timesOutAt <= deadline) continue; + + clearTimeout(task.timeout); + task.timesOutAt = deadline; + task.timeout = setTimeout( + async () => { + await this.handleTaskTimeout(taskId); + }, + Math.max(deadline - Date.now(), 0), + ); + cappedTaskIds.push(taskId); + } + + if (cappedTaskIds.length > 0) { + this.logger.info( + `Capped ${cappedTaskIds.length} in-flight task timeout(s) to fit the shutdown window (task IDs: ${cappedTaskIds.join(', ')})`, + ); + } + } + /** * For testing only */ diff --git a/packages/cli/src/task-runners/task-runner-process-base.ts b/packages/cli/src/task-runners/task-runner-process-base.ts index a2222891cb9..318f589347c 100644 --- a/packages/cli/src/task-runners/task-runner-process-base.ts +++ b/packages/cli/src/task-runners/task-runner-process-base.ts @@ -16,6 +16,9 @@ export type ChildProcess = ReturnType; const RESTART_RETRY_BASE_DELAY_MS = 5_000; const RESTART_RETRY_MAX_DELAY_MS = 30_000; +/** How long a stopping runner gets to exit gracefully before it is force-killed. */ +export const RUNNER_EXIT_GRACE_MS = 2_000; + export function restartRetryDelay(attempt: number): number { return Math.min(RESTART_RETRY_BASE_DELAY_MS * 2 ** (attempt - 1), RESTART_RETRY_MAX_DELAY_MS); } @@ -131,7 +134,13 @@ export abstract class TaskRunnerProcessBase extends TypedEmitter