fix(core): Handle task runner accept timeout error (#14709)

Author: ivov

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

@@ -1,13 +1,14 @@
 import type { TaskRunnersConfig } from '@n8n/config';
 import type { RunnerMessage, TaskResultData } from '@n8n/task-runner';
 import { mock } from 'jest-mock-extended';
+import type { Logger } from 'n8n-core';
 import { ApplicationError, type INodeTypeBaseDescription } from 'n8n-workflow';
 
 import { Time } from '@/constants';
 import type { TaskRunnerLifecycleEvents } from '@/task-runners/task-runner-lifecycle-events';
 
 import { TaskRejectError } from '../errors/task-reject.error';
-import { TaskRunnerTimeoutError } from '../errors/task-runner-timeout.error';
+import { TaskRunnerExecutionTimeoutError } from '../errors/task-runner-execution-timeout.error';
 import { TaskBroker } from '../task-broker.service';
 import type { TaskOffer, TaskRequest, TaskRunner } from '../task-broker.service';
 
@@ -715,7 +716,7 @@ describe('TaskBroker', () => {
 		});
 	});
 
-	describe('task timeouts', () => {
+	describe('task execution timeouts', () => {
 		let taskBroker: TaskBroker;
 		let config: TaskRunnersConfig;
 		let runnerLifecycleEvents = mock<TaskRunnerLifecycleEvents>();
@@ -879,11 +880,54 @@ describe('TaskBroker', () => {
 			expect(requesterCallback).toHaveBeenCalledWith({
 				type: 'broker:taskerror',
 				taskId,
-				error: expect.any(TaskRunnerTimeoutError),
+				error: expect.any(TaskRunnerExecutionTimeoutError),
 			});
 
 			expect(clearTimeout).toHaveBeenCalled();
 			expect(taskBroker.getTasks().get(taskId)).toBeUndefined();
 		});
 	});
+
+	describe('task runner accept timeout', () => {
+		it('broker should handle timeout when waiting for acknowledgment of offer accept', async () => {
+			const runnerId = 'runner1';
+			const runner = mock<TaskRunner>({ id: runnerId });
+			const messageCallback = jest.fn();
+			const loggerMock = mock<Logger>();
+
+			taskBroker = new TaskBroker(loggerMock, mock(), mock());
+			taskBroker.registerRunner(runner, messageCallback);
+
+			const offer: TaskOffer = {
+				offerId: 'offer1',
+				runnerId,
+				taskType: 'taskType1',
+				validFor: 1000,
+				validUntil: createValidUntil(1000),
+			};
+
+			const request: TaskRequest = {
+				requestId: 'request1',
+				requesterId: 'requester1',
+				taskType: 'taskType1',
+			};
+
+			jest.useFakeTimers();
+
+			const acceptPromise = taskBroker.acceptOffer(offer, request);
+
+			jest.advanceTimersByTime(2100);
+
+			await acceptPromise;
+
+			expect(request.acceptInProgress).toBe(false);
+			expect(loggerMock.warn).toHaveBeenCalledWith(
+				expect.stringContaining(
+					`Runner (${runnerId}) took too long to acknowledge acceptance of task`,
+				),
+			);
+
+			jest.useRealTimers();
+		});
+	});
 });

packages/cli/src/task-runners/task-broker/errors/task-runner-accept-timeout.error.ts

@@ -0,0 +1,9 @@
+import { OperationalError } from 'n8n-workflow';
+
+export class TaskRunnerAcceptTimeoutError extends OperationalError {
+	constructor(taskId: string, runnerId: string) {
+		super(`Runner (${runnerId}) took too long to acknowledge acceptance of task (${taskId})`, {
+			level: 'warning',
+		});
+	}
+}

packages/cli/src/task-runners/task-broker/errors/task-runner-timeout.error.ts renamed to packages/cli/src/task-runners/task-broker/errors/task-runner-execution-timeout.error.ts

@@ -1,7 +1,7 @@
 import type { TaskRunnerMode } from '@n8n/config/src/configs/runners.config';
 import { OperationalError } from 'n8n-workflow';
 
-export class TaskRunnerTimeoutError extends OperationalError {
+export class TaskRunnerExecutionTimeoutError extends OperationalError {
 	description: string;
 
 	constructor({

packages/cli/src/task-runners/task-broker/task-broker.service.ts

@@ -14,7 +14,8 @@ import config from '@/config';
 import { Time } from '@/constants';
 import { TaskDeferredError } from '@/task-runners/task-broker/errors/task-deferred.error';
 import { TaskRejectError } from '@/task-runners/task-broker/errors/task-reject.error';
-import { TaskRunnerTimeoutError } from '@/task-runners/task-broker/errors/task-runner-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 { TaskRunnerLifecycleEvents } from '@/task-runners/task-runner-lifecycle-events';
 
 export interface TaskRunner {
@@ -468,7 +469,7 @@ export class TaskBroker {
 
 		await this.taskErrorHandler(
 			taskId,
-			new TaskRunnerTimeoutError({
+			new TaskRunnerExecutionTimeoutError({
 				taskTimeout,
 				isSelfHosted: config.getEnv('deployment.type') !== 'cloud',
 				mode,
@@ -513,7 +514,7 @@ export class TaskBroker {
 
 				// TODO: customisable timeout
 				setTimeout(() => {
-					reject('Runner timed out');
+					reject(new TaskRunnerAcceptTimeoutError(taskId, offer.runnerId));
 				}, 2000);
 			});
 
@@ -535,6 +536,10 @@ export class TaskBroker {
 				this.pendingTaskRequests.push(request); // will settle on receiving task offer from runner
 				return;
 			}
+			if (e instanceof TaskRunnerAcceptTimeoutError) {
+				this.logger.warn(e.message);
+				return;
+			}
 			throw e;
 		}