fix(core): Avoid restarting task runners that are only slow (#36530)

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

@@ -1538,6 +1538,12 @@ describe('TaskBroker', () => {
 			await acceptPromise;
 		};
 
+		const trackTaskFor = (runnerId: string) => {
+			taskBroker.setTasks({
+				task1: { id: 'task1', runnerId, requesterId: 'requester1', taskType: 'taskType1' },
+			});
+		};
+
 		const answerAcceptance = async (respond: (taskId: string) => void) => {
 			const acceptPromise = taskBroker.acceptOffer(offerFrom('runner1'), requestFor());
 			const [taskId] = taskBroker.getRunnerAcceptRejects().keys();
@@ -1560,6 +1566,153 @@ describe('TaskBroker', () => {
 			expect(lifecycleEvents.emit).toHaveBeenCalledTimes(1);
 		});
 
+		it('should count acknowledgment timeouts firing within the same stall as a single timeout', async () => {
+			const acceptances = Array.from(
+				{ length: 3 },
+				async () => await taskBroker.acceptOffer(offerFrom('runner1'), requestFor()),
+			);
+			vi.advanceTimersByTime(ACCEPT_TIMEOUT_MS);
+			await Promise.all(acceptances);
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+
+			// the coalesced burst counted as one strike, so two spaced timeouts reach the threshold
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).toHaveBeenCalledTimes(1);
+			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
+				runnerId: 'runner1',
+			});
+		});
+
+		it('should reset the count when the runner completes a task', async () => {
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+			trackTaskFor('runner1');
+			await taskBroker.onRunnerMessage('runner1', {
+				type: 'runner:taskdone',
+				taskId: 'task1',
+				data: mock<TaskResultData>(),
+			});
+
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+		});
+
+		it('should not reset the count on a result for a task the runner was not given', async () => {
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+			trackTaskFor('runner2');
+			await taskBroker.onRunnerMessage('runner1', {
+				type: 'runner:taskdone',
+				taskId: 'task1',
+				data: mock<TaskResultData>(),
+			});
+
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
+				runnerId: 'runner1',
+			});
+		});
+
+		it('should not reset the count on a result for an untracked task', async () => {
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+			await taskBroker.onRunnerMessage('runner1', {
+				type: 'runner:taskdone',
+				taskId: 'unknown-task',
+				data: mock<TaskResultData>(),
+			});
+
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
+				runnerId: 'runner1',
+			});
+		});
+
+		it('should reset the count when the runner reports a task error', async () => {
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+			trackTaskFor('runner1');
+			await taskBroker.onRunnerMessage('runner1', {
+				type: 'runner:taskerror',
+				taskId: 'task1',
+				error: new Error('some error'),
+			});
+
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+		});
+
+		it('should not reset the count when the runner only keeps sending offers', async () => {
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+			await taskBroker.onRunnerMessage('runner1', {
+				type: 'runner:taskoffer',
+				taskType: 'taskType1',
+				offerId: 'offer2',
+				validFor: 10_000,
+			});
+
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).toHaveBeenCalledTimes(1);
+			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
+				runnerId: 'runner1',
+			});
+		});
+
+		it('should not report a runner that still has tasks in flight', async () => {
+			taskBroker.setTasks({
+				task1: {
+					id: 'task1',
+					runnerId: 'runner1',
+					requesterId: 'requester1',
+					taskType: 'taskType1',
+				},
+			});
+
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+		});
+
+		it('should report on the next timeout once the in-flight tasks are gone', async () => {
+			taskBroker.setTasks({
+				task1: {
+					id: 'task1',
+					runnerId: 'runner1',
+					requesterId: 'requester1',
+					taskType: 'taskType1',
+				},
+			});
+
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+
+			taskBroker.setTasks({});
+
+			// the threshold was already reached, so no further timeouts need to be re-earned
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).toHaveBeenCalledTimes(1);
+			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
+				runnerId: 'runner1',
+			});
+		});
+
 		it('should reset the count when the runner acknowledges an acceptance', async () => {
 			await timeOutAcceptance();
 			await timeOutAcceptance();
@@ -1606,17 +1759,37 @@ describe('TaskBroker', () => {
 			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
 		});
 
-		it('should report a runner that reached the threshold after deregistering', async () => {
-			const acceptances = Array.from(
-				{ length: 3 },
-				async () => await taskBroker.acceptOffer(offerFrom('runner1'), requestFor()),
-			);
+		it('should keep counting timeouts across a backward system clock jump', async () => {
+			await timeOutAcceptance();
+			await timeOutAcceptance();
 
-			// the transport deregisters the runner while the acceptances are still settling,
-			// so their timeouts are all counted against a runner no longer known
-			vi.advanceTimersByTime(ACCEPT_TIMEOUT_MS);
-			taskBroker.deregisterRunner('runner1', new Error('connection lost'));
-			await Promise.all(acceptances);
+			// a clock correction must not make the next timeout look like part of the last burst
+			vi.setSystemTime(Date.now() - 60_000);
+			await timeOutAcceptance();
+
+			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
+				runnerId: 'runner1',
+			});
+		});
+
+		it('should report a runner that keeps offering but never acknowledges', async () => {
+			taskBroker.registerRequester('requester1', vi.fn());
+			taskBroker.taskRequested({
+				requestId: 'request1',
+				requesterId: 'requester1',
+				taskType: 'taskType1',
+				timeout: taskBroker['createRequestTimeout']('request1'),
+			});
+
+			for (let n = 0; n < 3; n++) {
+				await taskBroker.onRunnerMessage('runner1', {
+					type: 'runner:taskoffer',
+					taskType: 'taskType1',
+					offerId: `offer${n}`,
+					validFor: 10_000,
+				});
+				await vi.advanceTimersByTimeAsync(ACCEPT_TIMEOUT_MS);
+			}
 
 			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
 				runnerId: 'runner1',
@@ -1642,6 +1815,7 @@ describe('TaskBroker', () => {
 
 	describe('silent runner detection', () => {
 		const REQUEST_TIMEOUT_MS = 60_000;
+		const MIN_SILENCE_DURATION_MS = 2_000;
 
 		let lifecycleEvents: MockProxy<TaskRunnerLifecycleEvents>;
 		let requesterCallback: ReturnType<typeof vi.fn<RequesterMessageCallback>>;
@@ -1671,20 +1845,25 @@ describe('TaskBroker', () => {
 			);
 		};
 
-		const letRequestExpire = () => {
+		const enqueueRequest = (requestId: string) => {
 			taskBroker.taskRequested({
-				requestId: 'request1',
+				requestId,
 				requesterId: 'requester1',
 				taskType: 'taskType1',
-				timeout: taskBroker['createRequestTimeout']('request1'),
+				timeout: taskBroker['createRequestTimeout'](requestId),
 			});
+		};
+
+		const letRequestExpire = (requestId = 'request1') => {
+			enqueueRequest(requestId);
 			vi.advanceTimersByTime(REQUEST_TIMEOUT_MS);
 		};
 
-		it('should report a reachable runner that sent no offers while a request expired', () => {
+		it('should report a reachable runner that stayed silent across two request expiries', () => {
 			registerRunner();
 
-			letRequestExpire();
+			letRequestExpire('request1');
+			letRequestExpire('request2');
 
 			expect(lifecycleEvents.emit).toHaveBeenCalledTimes(1);
 			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
@@ -1697,6 +1876,62 @@ describe('TaskBroker', () => {
 			});
 		});
 
+		it('should not report a runner observed silent at a single request expiry', () => {
+			registerRunner();
+
+			letRequestExpire('request1');
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+		});
+
+		it('should not report a runner when the expiries observing it are too close together', () => {
+			registerRunner();
+
+			enqueueRequest('request1');
+			enqueueRequest('request2');
+			vi.advanceTimersByTime(REQUEST_TIMEOUT_MS);
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+		});
+
+		it('should not report a runner when the expiries are just under the silence floor', () => {
+			registerRunner();
+
+			enqueueRequest('request1');
+			vi.advanceTimersByTime(MIN_SILENCE_DURATION_MS - 1);
+			enqueueRequest('request2');
+			vi.advanceTimersByTime(REQUEST_TIMEOUT_MS);
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+		});
+
+		it('should report a runner when the expiries are exactly the silence floor apart', () => {
+			registerRunner();
+
+			enqueueRequest('request1');
+			vi.advanceTimersByTime(MIN_SILENCE_DURATION_MS);
+			enqueueRequest('request2');
+			vi.advanceTimersByTime(REQUEST_TIMEOUT_MS);
+
+			expect(lifecycleEvents.emit).toHaveBeenCalledWith('runner:unresponsive', {
+				runnerId: 'runner1',
+			});
+		});
+
+		it('should not report a runner that sent a message between expiries', async () => {
+			registerRunner();
+
+			letRequestExpire('request1');
+			await taskBroker.onRunnerMessage('runner1', {
+				type: 'runner:taskdone',
+				taskId: 'task1',
+				data: mock<TaskResultData>(),
+			});
+			letRequestExpire('request2');
+
+			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
+		});
+
 		it('should not report a runner with an in-flight task', () => {
 			registerRunner();
 			taskBroker.setTasks({
@@ -1708,7 +1943,8 @@ describe('TaskBroker', () => {
 				},
 			});
 
-			letRequestExpire();
+			letRequestExpire('request1');
+			letRequestExpire('request2');
 
 			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
 		});
@@ -1725,7 +1961,8 @@ describe('TaskBroker', () => {
 				},
 			]);
 
-			letRequestExpire();
+			letRequestExpire('request1');
+			letRequestExpire('request2');
 
 			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
 		});
@@ -1742,7 +1979,8 @@ describe('TaskBroker', () => {
 				},
 			]);
 
-			letRequestExpire();
+			letRequestExpire('request1');
+			letRequestExpire('request2');
 
 			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
 		});
@@ -1753,7 +1991,8 @@ describe('TaskBroker', () => {
 				task1: { accept: vi.fn(), reject: vi.fn(), runnerId: 'runner1' },
 			});
 
-			letRequestExpire();
+			letRequestExpire('request1');
+			letRequestExpire('request2');
 
 			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
 		});
@@ -1761,7 +2000,8 @@ describe('TaskBroker', () => {
 		it('should not report an unreachable runner', () => {
 			registerRunner(() => false);
 
-			letRequestExpire();
+			letRequestExpire('request1');
+			letRequestExpire('request2');
 
 			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
 		});
@@ -1769,13 +2009,14 @@ describe('TaskBroker', () => {
 		it('should not report a runner that does not support the task type', () => {
 			taskBroker.registerRunner(mock<TaskRunner>({ id: 'runner1', taskTypes: ['other'] }), vi.fn());
 
-			letRequestExpire();
+			letRequestExpire('request1');
+			letRequestExpire('request2');
 
 			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
 		});
 
 		it('should expire the request without reporting when no runner is registered', () => {
-			letRequestExpire();
+			letRequestExpire('request1');
 
 			expect(lifecycleEvents.emit).not.toHaveBeenCalled();
 			expect(requesterCallback).toHaveBeenCalledWith({

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

@@ -48,6 +48,14 @@ const MAX_REQUEST_TIMEOUT_REFRESHES = 3;
 
 const MAX_CONSECUTIVE_ACCEPT_TIMEOUTS = 3;
 
+/**
+ * How far apart two silence observations must be to count as two.
+ * Requests pending together expire in the same tick, so without this floor a single
+ * instant of silence would satisfy the two-observation rule and could catch a runner
+ * that is only briefly offerless between tasks.
+ */
+const MIN_SILENCE_DURATION_MS = 2_000;
+
 export interface TaskRequest {
 	requestId: string;
 	requesterId: string;
@@ -100,7 +108,13 @@ export class TaskBroker {
 		{ accept: RequesterAcceptCallback; reject: TaskRejectCallback }
 	> = new Map();
 
-	private consecutiveAcceptTimeouts: Map<TaskRunner['id'], number> = new Map();
+	private consecutiveAcceptTimeouts: Map<
+		TaskRunner['id'],
+		{ count: number; lastTimeoutAt: number }
+	> = new Map();
+
+	/** When each runner was first observed silent, cleared on any message from it. */
+	private silentRunnersSince: Map<TaskRunner['id'], number> = new Map();
 
 	private pendingTaskOffers: TaskOffer[] = [];
 
@@ -215,6 +229,7 @@ export class TaskBroker {
 	deregisterRunner(runnerId: string, error: Error) {
 		this.knownRunners.delete(runnerId);
 		this.consecutiveAcceptTimeouts.delete(runnerId);
+		this.silentRunnersSince.delete(runnerId);
 
 		this.discardOffersFrom(runnerId);
 
@@ -263,6 +278,10 @@ export class TaskBroker {
 		if (!runner) {
 			return;
 		}
+
+		// Any message from the runner disproves the silence a report would be based on.
+		this.silentRunnersSince.delete(runnerId);
+
 		switch (message.type) {
 			case 'runner:taskaccepted':
 				this.handleRunnerAccept(message.taskId);
@@ -286,9 +305,11 @@ export class TaskBroker {
 				});
 				break;
 			case 'runner:taskdone':
+				this.resetAcceptTimeoutsOnOwnTask(runnerId, message.taskId);
 				await this.taskDoneHandler(message.taskId, message.data);
 				break;
 			case 'runner:taskerror':
+				this.resetAcceptTimeoutsOnOwnTask(runnerId, message.taskId);
 				await this.taskErrorHandler(message.taskId, message.error);
 				break;
 			case 'runner:taskdatarequest':
@@ -767,44 +788,82 @@ export class TaskBroker {
 	}
 
 	/**
-	 * Counts consecutive acknowledgement timeouts per runner.
-	 * Any application-level reply (accept, reject, defer) proves the channel is alive and resets the count.
-	 * At the threshold, the runner is reported unresponsive exactly once,
-	 * so its transport can be torn down and the runner restarted.
+	 * Clears acknowledgement strikes when a runner reports on a task the broker tracks for
+	 * it, which proves its accept path is alive. A result for a task the broker no longer
+	 * tracks, or never assigned to this runner, proves nothing and is ignored.
+	 */
+	private resetAcceptTimeoutsOnOwnTask(runnerId: TaskRunner['id'], taskId: Task['id']) {
+		if (this.tasks.get(taskId)?.runnerId === runnerId) {
+			this.consecutiveAcceptTimeouts.delete(runnerId);
+		}
+	}
+
+	/**
+	 * Counts consecutive acknowledgement timeouts per runner and reports it unresponsive
+	 * once the threshold is reached. Timeouts within one acceptance window of the previous
+	 * one count as a single timeout, and any acknowledgement or task result resets the
+	 * count. Task offers do not reset it, since an alive offer loop with a dead accept path
+	 * is the state this detects.
 	 */
 	private flagRunnerIfUnresponsive(runnerId: TaskRunner['id']) {
-		const failures = (this.consecutiveAcceptTimeouts.get(runnerId) ?? 0) + 1;
+		const now = this.monotonicNowMs();
+		const previous = this.consecutiveAcceptTimeouts.get(runnerId);
+		const acceptWindowMs = this.taskRunnersConfig.taskAcceptTimeout * Time.seconds.toMilliseconds;
 
-		if (failures < MAX_CONSECUTIVE_ACCEPT_TIMEOUTS) {
-			this.consecutiveAcceptTimeouts.set(runnerId, failures);
-		} else {
+		if (previous && now - previous.lastTimeoutAt < acceptWindowMs) {
+			return;
+		}
+
+		const failures = Math.min((previous?.count ?? 0) + 1, MAX_CONSECUTIVE_ACCEPT_TIMEOUTS);
+
+		this.consecutiveAcceptTimeouts.set(runnerId, { count: failures, lastTimeoutAt: now });
+
+		if (failures < MAX_CONSECUTIVE_ACCEPT_TIMEOUTS) return;
+
+		const reported = this.reportUnresponsive(
+			runnerId,
+			`failed to acknowledge ${MAX_CONSECUTIVE_ACCEPT_TIMEOUTS} consecutive task acceptances`,
+		);
+
+		if (reported) {
 			this.consecutiveAcceptTimeouts.delete(runnerId);
-			this.reportUnresponsive(
-				runnerId,
-				`failed to acknowledge ${MAX_CONSECUTIVE_ACCEPT_TIMEOUTS} consecutive task acceptances`,
-			);
 		}
 	}
 
 	/**
 	 * Reports as unresponsive every reachable runner for `taskType` with no sign of life:
-	 * no pending offer, no in-flight task, no acceptance in progress.
-	 *
-	 * A healthy runner with spare capacity keeps offers pending,
-	 * so a request expiring next to a silent runner means the runner's offer loop has stalled
-	 * and its transport should be torn down so the runner can be restarted.
-	 *
-	 * A no-op when no runner is registered, which is normal while a runner is still starting up.
+	 * no pending offer, no in-flight task, no acceptance in progress, no message received.
+	 * Since a healthy runner is briefly offerless between tasks, a runner is only reported
+	 * once observed silent at two request expiries at least the minimum silence duration apart.
 	 */
 	private flagSilentRunners(taskType: string) {
 		this.expireTasks();
 
+		const now = this.monotonicNowMs();
+
 		[...this.knownRunners.values()]
 			.filter(({ runner }) => runner.taskTypes.includes(taskType))
 			.filter(({ isRunnerReachable }) => isRunnerReachable())
-			.filter(({ runner }) => this.isSilent(runner.id))
 			.forEach(({ runner }) => {
-				this.reportUnresponsive(runner.id, 'sent no task offers while a task request expired');
+				if (!this.isSilent(runner.id)) {
+					this.silentRunnersSince.delete(runner.id);
+					return;
+				}
+
+				const silentSince = this.silentRunnersSince.get(runner.id);
+
+				if (silentSince === undefined) {
+					this.silentRunnersSince.set(runner.id, now);
+				} else if (now - silentSince >= MIN_SILENCE_DURATION_MS) {
+					const reported = this.reportUnresponsive(
+						runner.id,
+						'sent no task offers while task requests expired',
+					);
+
+					if (reported) {
+						this.silentRunnersSince.delete(runner.id);
+					}
+				}
 			});
 	}
 
@@ -812,13 +871,33 @@ export class TaskBroker {
 	 * Reports a runner as unresponsive, so its transport can be torn down and, in internal
 	 * mode, its process force-restarted.
 	 *
-	 * Reports a runner that is no longer registered too: concurrent acceptances can reach the
-	 * timeout threshold after the transport deregistered the runner, and a process that
-	 * outlived its transport is exactly what still needs restarting.
+	 * Skipped while the runner has in-flight tasks: tearing it down would fail tasks that
+	 * may still complete. A stuck runner is still recovered once those tasks hit their own
+	 * execution timeout, which force-restarts the process in internal mode and, in external
+	 * mode, untracks the tasks so the next report is no longer skipped.
+	 *
+	 * @returns whether the runner was reported.
 	 */
-	private reportUnresponsive(runnerId: TaskRunner['id'], cause: string) {
+	private reportUnresponsive(runnerId: TaskRunner['id'], cause: string): boolean {
+		if (this.getInFlightTaskIds(runnerId).length > 0) {
+			this.logger.debug(
+				`Runner (${runnerId}) ${cause}, but it still has tasks in flight, so not reporting it as unresponsive`,
+			);
+			return false;
+		}
+
 		this.logger.warn(`Runner (${runnerId}) ${cause}, reporting it as unresponsive`);
 		this.taskRunnerLifecycleEvents.emit('runner:unresponsive', { runnerId });
+
+		return true;
+	}
+
+	/**
+	 * Milliseconds from a monotonic clock, so a system clock adjustment cannot make two
+	 * observations look closer together or further apart than they were.
+	 */
+	private monotonicNowMs() {
+		return Number(process.hrtime.bigint() / 1_000_000n);
 	}
 
 	private isSilent(runnerId: TaskRunner['id']) {