feat(core): Implement task timeouts and heartbeats for runners (no-changelog) (#11690)

Co-authored-by: कारतोफ्फेलस्क्रिप्ट™ <[email protected]>

Author: ivov

packages/@n8n/config/src/configs/runners.config.ts

@@ -53,4 +53,12 @@ export class TaskRunnersConfig {
 	/** Should the output of deduplication be asserted for correctness */
 	@Env('N8N_RUNNERS_ASSERT_DEDUPLICATION_OUTPUT')
 	assertDeduplicationOutput: boolean = false;
+
+	/** How long (in seconds) a task is allowed to take for completion, else the task will be aborted and the runner restarted. Must be greater than 0. */
+	@Env('N8N_RUNNERS_TASK_TIMEOUT')
+	taskTimeout: number = 60;
+
+	/** How often (in seconds) the runner must send a heartbeat to the broker, else the task will be aborted and the runner restarted. Must be greater than 0. */
+	@Env('N8N_RUNNERS_HEARTBEAT_INTERVAL')
+	heartbeatInterval: number = 30;
 }

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

@@ -234,6 +234,8 @@ describe('GlobalConfig', () => {
 			maxOldSpaceSize: '',
 			maxConcurrency: 5,
 			assertDeduplicationOutput: false,
+			taskTimeout: 60,
+			heartbeatInterval: 30,
 		},
 		sentry: {
 			backendDsn: '',

packages/@n8n/task-runner/src/config/base-runner-config.ts

@@ -1,4 +1,16 @@
-import { Config, Env } from '@n8n/config';
+import { Config, Env, Nested } from '@n8n/config';
+
+@Config
+class HealthcheckServerConfig {
+	@Env('N8N_RUNNERS_SERVER_ENABLED')
+	enabled: boolean = false;
+
+	@Env('N8N_RUNNERS_SERVER_HOST')
+	host: string = '127.0.0.1';
+
+	@Env('N8N_RUNNERS_SERVER_PORT')
+	port: number = 5680;
+}
 
 @Config
 export class BaseRunnerConfig {
@@ -13,4 +25,7 @@ export class BaseRunnerConfig {
 
 	@Env('N8N_RUNNERS_MAX_CONCURRENCY')
 	maxConcurrency: number = 5;
+
+	@Nested
+	healthcheckServer!: HealthcheckServerConfig;
 }

packages/@n8n/task-runner/src/healthcheck-server.ts

@@ -0,0 +1,38 @@
+import { ApplicationError } from 'n8n-workflow';
+import { createServer } from 'node:http';
+
+export class HealthcheckServer {
+	private server = createServer((_, res) => {
+		res.writeHead(200);
+		res.end('OK');
+	});
+
+	async start(host: string, port: number) {
+		return await new Promise<void>((resolve, reject) => {
+			const portInUseErrorHandler = (error: NodeJS.ErrnoException) => {
+				if (error.code === 'EADDRINUSE') {
+					reject(new ApplicationError(`Port ${port} is already in use`));
+				} else {
+					reject(error);
+				}
+			};
+
+			this.server.on('error', portInUseErrorHandler);
+
+			this.server.listen(port, host, () => {
+				this.server.removeListener('error', portInUseErrorHandler);
+				console.log(`Healthcheck server listening on ${host}, port ${port}`);
+				resolve();
+			});
+		});
+	}
+
+	async stop() {
+		return await new Promise<void>((resolve, reject) => {
+			this.server.close((error) => {
+				if (error) reject(error);
+				else resolve();
+			});
+		});
+	}
+}

packages/@n8n/task-runner/src/start.ts

@@ -3,8 +3,10 @@ import Container from 'typedi';
 
 import { MainConfig } from './config/main-config';
 import type { ErrorReporter } from './error-reporter';
+import type { HealthcheckServer } from './healthcheck-server';
 import { JsTaskRunner } from './js-task-runner/js-task-runner';
 
+let healthcheckServer: HealthcheckServer | undefined;
 let runner: JsTaskRunner | undefined;
 let isShuttingDown = false;
 let errorReporter: ErrorReporter | undefined;
@@ -22,6 +24,7 @@ function createSignalHandler(signal: string) {
 			if (runner) {
 				await runner.stop();
 				runner = undefined;
+				void healthcheckServer?.stop();
 			}
 
 			if (errorReporter) {
@@ -49,6 +52,14 @@ void (async function start() {
 
 	runner = new JsTaskRunner(config);
 
+	const { enabled, host, port } = config.baseRunnerConfig.healthcheckServer;
+
+	if (enabled) {
+		const { HealthcheckServer } = await import('./healthcheck-server');
+		healthcheckServer = new HealthcheckServer();
+		await healthcheckServer.start(host, port);
+	}
+
 	process.on('SIGINT', createSignalHandler('SIGINT'));
 	process.on('SIGTERM', createSignalHandler('SIGTERM'));
 })().catch((e) => {

packages/cli/src/runners/__tests__/task-broker.test.ts

@@ -1,8 +1,12 @@
+import type { TaskRunnersConfig } from '@n8n/config';
 import type { RunnerMessage, TaskResultData } from '@n8n/task-runner';
 import { mock } from 'jest-mock-extended';
-import type { INodeTypeBaseDescription } from 'n8n-workflow';
+import { ApplicationError, type INodeTypeBaseDescription } from 'n8n-workflow';
+
+import { Time } from '@/constants';
 
 import { TaskRejectError } from '../errors';
+import type { RunnerLifecycleEvents } from '../runner-lifecycle-events';
 import { TaskBroker } from '../task-broker.service';
 import type { TaskOffer, TaskRequest, TaskRunner } from '../task-broker.service';
 
@@ -12,7 +16,7 @@ describe('TaskBroker', () => {
 	let taskBroker: TaskBroker;
 
 	beforeEach(() => {
-		taskBroker = new TaskBroker(mock());
+		taskBroker = new TaskBroker(mock(), mock(), mock());
 		jest.restoreAllMocks();
 	});
 
@@ -618,4 +622,131 @@ describe('TaskBroker', () => {
 			});
 		});
 	});
+
+	describe('task timeouts', () => {
+		let taskBroker: TaskBroker;
+		let config: TaskRunnersConfig;
+		let runnerLifecycleEvents = mock<RunnerLifecycleEvents>();
+
+		beforeAll(() => {
+			jest.useFakeTimers();
+			config = mock<TaskRunnersConfig>({ taskTimeout: 30 });
+			taskBroker = new TaskBroker(mock(), config, runnerLifecycleEvents);
+		});
+
+		afterAll(() => {
+			jest.useRealTimers();
+		});
+
+		it('on sending task, we should set up task timeout', async () => {
+			jest.spyOn(global, 'setTimeout');
+
+			const taskId = 'task1';
+			const runnerId = 'runner1';
+			const runner = mock<TaskRunner>({ id: runnerId });
+			const runnerMessageCallback = jest.fn();
+
+			taskBroker.registerRunner(runner, runnerMessageCallback);
+			taskBroker.setTasks({
+				[taskId]: { id: taskId, runnerId, requesterId: 'requester1', taskType: 'test' },
+			});
+
+			await taskBroker.sendTaskSettings(taskId, {});
+
+			expect(setTimeout).toHaveBeenCalledWith(
+				expect.any(Function),
+				config.taskTimeout * Time.seconds.toMilliseconds,
+			);
+		});
+
+		it('on task completion, we should clear timeout', async () => {
+			jest.spyOn(global, 'clearTimeout');
+
+			const taskId = 'task1';
+			const runnerId = 'runner1';
+			const requesterId = 'requester1';
+			const requesterCallback = jest.fn();
+
+			taskBroker.registerRequester(requesterId, requesterCallback);
+			taskBroker.setTasks({
+				[taskId]: {
+					id: taskId,
+					runnerId,
+					requesterId,
+					taskType: 'test',
+					timeout: setTimeout(() => {}, config.taskTimeout * Time.seconds.toMilliseconds),
+				},
+			});
+
+			await taskBroker.taskDoneHandler(taskId, { result: [] });
+
+			expect(clearTimeout).toHaveBeenCalled();
+			expect(taskBroker.getTasks().get(taskId)).toBeUndefined();
+		});
+
+		it('on task error, we should clear timeout', async () => {
+			jest.spyOn(global, 'clearTimeout');
+
+			const taskId = 'task1';
+			const runnerId = 'runner1';
+			const requesterId = 'requester1';
+			const requesterCallback = jest.fn();
+
+			taskBroker.registerRequester(requesterId, requesterCallback);
+			taskBroker.setTasks({
+				[taskId]: {
+					id: taskId,
+					runnerId,
+					requesterId,
+					taskType: 'test',
+					timeout: setTimeout(() => {}, config.taskTimeout * Time.seconds.toMilliseconds),
+				},
+			});
+
+			await taskBroker.taskErrorHandler(taskId, new Error('Test error'));
+
+			expect(clearTimeout).toHaveBeenCalled();
+			expect(taskBroker.getTasks().get(taskId)).toBeUndefined();
+		});
+
+		it('on timeout, we should emit `runner:timed-out-during-task` event and send error to requester', async () => {
+			jest.spyOn(global, 'clearTimeout');
+
+			const taskId = 'task1';
+			const runnerId = 'runner1';
+			const requesterId = 'requester1';
+			const runner = mock<TaskRunner>({ id: runnerId });
+			const runnerCallback = jest.fn();
+			const requesterCallback = jest.fn();
+
+			taskBroker.registerRunner(runner, runnerCallback);
+			taskBroker.registerRequester(requesterId, requesterCallback);
+
+			taskBroker.setTasks({
+				[taskId]: { id: taskId, runnerId, requesterId, taskType: 'test' },
+			});
+
+			await taskBroker.sendTaskSettings(taskId, {});
+
+			jest.runAllTimers();
+
+			await Promise.resolve();
+
+			expect(runnerLifecycleEvents.emit).toHaveBeenCalledWith('runner:timed-out-during-task');
+
+			await Promise.resolve();
+
+			expect(clearTimeout).toHaveBeenCalled();
+
+			expect(requesterCallback).toHaveBeenCalledWith({
+				type: 'broker:taskerror',
+				taskId,
+				error: new ApplicationError(`Task execution timed out after ${config.taskTimeout} seconds`),
+			});
+
+			await Promise.resolve();
+
+			expect(taskBroker.getTasks().get(taskId)).toBeUndefined();
+		});
+	});
 });

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

@@ -7,6 +7,8 @@ import type { TaskRunnerAuthService } from '@/runners/auth/task-runner-auth.serv
 import { TaskRunnerProcess } from '@/runners/task-runner-process';
 import { mockInstance } from '@test/mocking';
 
+import type { RunnerLifecycleEvents } from '../runner-lifecycle-events';
+
 const spawnMock = jest.fn(() =>
 	mock<ChildProcess>({
 		stdout: {
@@ -25,7 +27,7 @@ describe('TaskRunnerProcess', () => {
 	runnerConfig.enabled = true;
 	runnerConfig.mode = 'internal_childprocess';
 	const authService = mock<TaskRunnerAuthService>();
-	let taskRunnerProcess = new TaskRunnerProcess(logger, runnerConfig, authService);
+	let taskRunnerProcess = new TaskRunnerProcess(logger, runnerConfig, authService, mock());
 
 	afterEach(async () => {
 		spawnMock.mockClear();
@@ -35,15 +37,35 @@ describe('TaskRunnerProcess', () => {
 		it('should throw if runner mode is external', () => {
 			runnerConfig.mode = 'external';
 
-			expect(() => new TaskRunnerProcess(logger, runnerConfig, authService)).toThrow();
+			expect(() => new TaskRunnerProcess(logger, runnerConfig, authService, mock())).toThrow();
 
 			runnerConfig.mode = 'internal_childprocess';
 		});
+
+		it('should register listener for `runner:failed-heartbeat-check` event', () => {
+			const runnerLifecycleEvents = mock<RunnerLifecycleEvents>();
+			new TaskRunnerProcess(logger, runnerConfig, authService, runnerLifecycleEvents);
+
+			expect(runnerLifecycleEvents.on).toHaveBeenCalledWith(
+				'runner:failed-heartbeat-check',
+				expect.any(Function),
+			);
+		});
+
+		it('should register listener for `runner:timed-out-during-task` event', () => {
+			const runnerLifecycleEvents = mock<RunnerLifecycleEvents>();
+			new TaskRunnerProcess(logger, runnerConfig, authService, runnerLifecycleEvents);
+
+			expect(runnerLifecycleEvents.on).toHaveBeenCalledWith(
+				'runner:timed-out-during-task',
+				expect.any(Function),
+			);
+		});
 	});
 
 	describe('start', () => {
 		beforeEach(() => {
-			taskRunnerProcess = new TaskRunnerProcess(logger, runnerConfig, authService);
+			taskRunnerProcess = new TaskRunnerProcess(logger, runnerConfig, authService, mock());
 		});
 
 		test.each([

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

@@ -0,0 +1,45 @@
+import type { TaskRunnersConfig } from '@n8n/config';
+import { mock } from 'jest-mock-extended';
+
+import { Time } from '@/constants';
+import { TaskRunnerWsServer } from '@/runners/runner-ws-server';
+
+describe('TaskRunnerWsServer', () => {
+	describe('heartbeat timer', () => {
+		it('should set up heartbeat timer on server start', async () => {
+			const setIntervalSpy = jest.spyOn(global, 'setInterval');
+
+			const server = new TaskRunnerWsServer(
+				mock(),
+				mock(),
+				mock(),
+				mock<TaskRunnersConfig>({ path: '/runners', heartbeatInterval: 30 }),
+				mock(),
+			);
+
+			expect(setIntervalSpy).toHaveBeenCalledWith(
+				expect.any(Function),
+				30 * Time.seconds.toMilliseconds,
+			);
+
+			await server.shutdown();
+		});
+
+		it('should clear heartbeat timer on server stop', async () => {
+			jest.spyOn(global, 'setInterval');
+			const clearIntervalSpy = jest.spyOn(global, 'clearInterval');
+
+			const server = new TaskRunnerWsServer(
+				mock(),
+				mock(),
+				mock(),
+				mock<TaskRunnersConfig>({ path: '/runners', heartbeatInterval: 30 }),
+				mock(),
+			);
+
+			await server.shutdown();
+
+			expect(clearIntervalSpy).toHaveBeenCalled();
+		});
+	});
+});