From 222eec5ccf7c70ad047f18633ae89df9e62f413f Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Wed, 19 Aug 2026 12:29:16 +0000 Subject: [PATCH] fix(core): Avoid restarting task runners that are only slow (#36530) --- .../__tests__/task-broker.service.test.ts | 285 ++++++++++++++++-- .../task-broker/task-broker.service.ts | 131 ++++++-- 2 files changed, 368 insertions(+), 48 deletions(-) diff --git a/packages/cli/src/task-runners/task-broker/__tests__/task-broker.service.test.ts b/packages/cli/src/task-runners/task-broker/__tests__/task-broker.service.test.ts index 48d344d0e4c..4afa320f7a5 100644 --- a/packages/cli/src/task-runners/task-broker/__tests__/task-broker.service.test.ts +++ b/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(), + }); + + 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(), + }); + + 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(), + }); + + 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; let requesterCallback: ReturnType>; @@ -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(), + }); + 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({ 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({ diff --git a/packages/cli/src/task-runners/task-broker/task-broker.service.ts b/packages/cli/src/task-runners/task-broker/task-broker.service.ts index 6dbf8b5cd75..be0b8623029 100644 --- a/packages/cli/src/task-runners/task-broker/task-broker.service.ts +++ b/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 = 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 = 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']) {