fix(core): Cap task runner task timeouts to the graceful shutdown window (backport to release-candidate/2.38.x) (#37765)

Co-authored-by: Mike Repeć <mike.repec@n8n.io>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

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

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

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<TaskRunnerLifecycleEvents>;
+
+		const taskId = 'task1';
+		const runnerId = 'runner1';
+		const requesterId = 'requester1';
+		let requesterCallback: ReturnType<typeof vi.fn<RequesterMessageCallback>>;
+
+		const setupTaskWithArmedTimer = async () => {
+			taskBroker.registerRunner(mock<TaskRunner>({ 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<TaskRunnerLifecycleEvents>();
+			requesterCallback = vi.fn<RequesterMessageCallback>();
+			config = mock<TaskRunnersConfig>({
+				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<TaskRunner>({ 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<TaskRunnersConfig>({
+				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;
 

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

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;
 

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<Task['id']> = [];
+
+		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
 	 */

packages/cli/src/task-runners/task-runner-process-base.ts

@@ -16,6 +16,9 @@ export type ChildProcess = ReturnType<typeof spawn>;
 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<TaskRunnerProce
 
 		if (this.process) {
 			this.process.kill();
-			await this._runPromise;
+			// A runner wedged on an in-flight task never exits gracefully and would
+			// hold shutdown until the force-kill timer. Escalate after a short grace.
+			await Promise.race([this._runPromise, sleep(RUNNER_EXIT_GRACE_MS)]);
+			if (this.process) {
+				this.process.kill('SIGKILL');
+				await this._runPromise;
+			}
 		}
 	}