fix(core): Stop task broker before task runner processes on shutdown (#37833)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: Flexicon

packages/cli/src/task-runners/__tests__/task-runner-module.test.ts

@@ -0,0 +1,53 @@
+import type { Logger } from '@n8n/backend-common';
+import type { TaskRunnersConfig } from '@n8n/config';
+import type { ErrorReporter } from 'n8n-core';
+import { mock } from 'vitest-mock-extended';
+
+import type { EventService } from '@/events/event.service';
+import type { TaskBrokerServer } from '@/task-runners/task-broker/task-broker-server';
+import type { JsTaskRunnerProcess } from '@/task-runners/task-runner-process-js';
+import type { PyTaskRunnerProcess } from '@/task-runners/task-runner-process-py';
+import { TaskRunnerModule } from '@/task-runners/task-runner-module';
+
+describe('TaskRunnerModule', () => {
+	describe('stop', () => {
+		it('should stop the broker server before the runner processes, so the drain can finish in-flight tasks first', async () => {
+			const logger = mock<Logger>();
+			logger.scoped.mockReturnValue(logger);
+			const module = new TaskRunnerModule(
+				logger,
+				mock<ErrorReporter>(),
+				mock<TaskRunnersConfig>(),
+				mock<EventService>(),
+			);
+
+			const brokerServer = mock<TaskBrokerServer>();
+			const jsRunnerProcess = mock<JsTaskRunnerProcess>();
+			const pyRunnerProcess = mock<PyTaskRunnerProcess>();
+			Object.assign(module, {
+				taskBrokerHttpServer: brokerServer,
+				jsRunnerProcess,
+				pyRunnerProcess,
+			});
+
+			const order: string[] = [];
+			// The broker stop completes asynchronously, so a stop() that merely starts
+			// it without awaiting would record the runner stops first.
+			brokerServer.stop.mockImplementation(async () => {
+				await new Promise((resolve) => setImmediate(resolve));
+				order.push('brokerServer');
+			});
+			jsRunnerProcess.stop.mockImplementation(async () => {
+				order.push('jsRunnerProcess');
+			});
+			pyRunnerProcess.stop.mockImplementation(async () => {
+				order.push('pyRunnerProcess');
+			});
+
+			await module.stop();
+
+			expect(order).toHaveLength(3);
+			expect(order[0]).toBe('brokerServer');
+		});
+	});
+});

packages/cli/src/task-runners/task-runner-module.ts

@@ -67,6 +67,14 @@ export class TaskRunnerModule {
 
 	@OnShutdown()
 	async stop() {
+		// Stop the broker server first: its drain lets in-flight tasks finish, so the
+		// runner processes are idle by the time they are stopped and exit within the
+		// short grace before the SIGKILL escalation.
+		if (this.taskBrokerHttpServer) {
+			await this.taskBrokerHttpServer.stop();
+			this.taskBrokerHttpServer = undefined;
+		}
+
 		const stopRunnerProcessTask = (async () => {
 			if (this.jsRunnerProcess) {
 				await this.jsRunnerProcess.stop();
@@ -81,14 +89,7 @@ export class TaskRunnerModule {
 			}
 		})();
 
-		const stopRunnerServerTask = (async () => {
-			if (this.taskBrokerHttpServer) {
-				await this.taskBrokerHttpServer.stop();
-				this.taskBrokerHttpServer = undefined;
-			}
-		})();
-
-		await Promise.all([stopRunnerProcessTask, stopPythonRunnerProcessTask, stopRunnerServerTask]);
+		await Promise.all([stopRunnerProcessTask, stopPythonRunnerProcessTask]);
 	}
 
 	private async loadTaskRequester() {