From 5cda774942bd33e08e2bc3af4474b952e5ba08e7 Mon Sep 17 00:00:00 2001 From: mojaza Date: Sat, 22 Aug 2026 01:28:17 -0700 Subject: [PATCH 1/4] [rush-daemon][WS2.9] Merge shared builds Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...shared-build-merging_2026-08-22-00-11.json | 11 + libraries/rush-daemon/README.md | 10 +- .../src/PhasedRequestEventMultiplexer.ts | 35 +- .../rush-daemon/src/PhasedRequestEventSink.ts | 8 +- .../rush-daemon/src/PhasedRequestRouter.ts | 629 ++++++++++++++---- .../src/test/PhasedRequestBatching.test.ts | 443 ++++++++++++ .../test/PhasedRequestRouterTestUtilities.ts | 6 +- .../test/RequestAdmissionIntegration.test.ts | 90 ++- 8 files changed, 1056 insertions(+), 176 deletions(-) create mode 100644 common/changes/@rushstack/rush-daemon/mojazayeri-shared-build-merging_2026-08-22-00-11.json create mode 100644 libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts diff --git a/common/changes/@rushstack/rush-daemon/mojazayeri-shared-build-merging_2026-08-22-00-11.json b/common/changes/@rushstack/rush-daemon/mojazayeri-shared-build-merging_2026-08-22-00-11.json new file mode 100644 index 0000000000..a1ad333e8b --- /dev/null +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-shared-build-merging_2026-08-22-00-11.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon", + "comment": "Merge compatible shared-build requests into one warm operation-graph iteration.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index 91f42d52c6..771c17db49 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -42,6 +42,12 @@ backpressured, ordered callbacks, followed exactly once by a typed final command drains. The result translates only that client's operation subset to Rush's success, warning, failure, or abort exit semantics. Warning-only builds honor the operation's configured `allowWarningsInSuccessfulBuild` state plus the request's immutable `RUSH_ALLOW_WARNINGS_IN_SUCCESSFUL_BUILD` environment override without mutating `process.env`. +Compatible phased `SHARED-BUILD` requests admitted before the next graph iteration starts are coalesced at a +deterministic event-loop-turn boundary. The router reconciles retained invalidations once, unions the clients' enabled +dependency closures, and schedules one iteration. Shared operations execute once, while each client subscribes only +to its own closure and derives its final result only from that subset. Requests admitted after scheduling starts form +a later batch. Cancelling or disconnecting one client removes its subscription without aborting work needed by other +clients; the graph iteration is aborted only after every client in that batch has stopped needing it. This layer deliberately does not reconstruct `PhasedScriptAction` command/plugin initialization. The typed phased request contract begins after an integration has produced a validated selection for the exact warm engine shape; @@ -74,6 +80,4 @@ Terminal width remains the immutable request-start value established by WS2.5. T rendering, so this layer does not forward `SIGWINCH`. Commands declaring a real controlling-terminal requirement receive a typed `requiresInProcess` policy result and are not executed by rushd; no pseudo-terminal is allocated or emulated. The future WS4 client will perform the actual in-process fallback and parse `--no-wait` / -`--wait-timeout`. Compatible `SHARED-BUILD` requests may hold admission leases concurrently, but the phased router -continues to serialize mutation of the single warm graph. WS2.9 will replace that internal graph lock with coordinated -selection merging. +`--wait-timeout`. diff --git a/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts b/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts index 431d476cc8..a6271bfaed 100644 --- a/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts +++ b/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts @@ -11,31 +11,28 @@ import type { ITerminalChunk } from '@rushstack/terminal'; export class PhasedRequestEventMultiplexer implements _IOperationGraphEventSink { readonly #workspaceSink: _IOperationGraphEventSink | undefined; - #requestSink: _IOperationGraphEventSink | undefined; + readonly #requestSinks: Set<_IOperationGraphEventSink> = new Set(); public constructor(workspaceSink: _IOperationGraphEventSink | undefined) { this.#workspaceSink = workspaceSink; } public subscribe(requestSink: _IOperationGraphEventSink): () => void { - if (this.#requestSink) { - throw new Error('A phased request event subscription is already active.'); - } - this.#requestSink = requestSink; + this.#requestSinks.add(requestSink); let subscribed: boolean = true; return () => { if (subscribed) { subscribed = false; - if (this.#requestSink === requestSink) { - this.#requestSink = undefined; - } + this.#requestSinks.delete(requestSink); } }; } public onOperationRegistered(operationId: string, silent: boolean): void { this.#workspaceSink?.onOperationRegistered?.(operationId, silent); - this.#requestSink?.onOperationRegistered?.(operationId, silent); + for (const requestSink of this.#requestSinks) { + requestSink.onOperationRegistered?.(operationId, silent); + } } public onOperationStatusChanged( @@ -43,26 +40,36 @@ export class PhasedRequestEventMultiplexer implements _IOperationGraphEventSink previousStatus: OperationStatus ): void { this.#workspaceSink?.onOperationStatusChanged?.(result, previousStatus); - this.#requestSink?.onOperationStatusChanged?.(result, previousStatus); + for (const requestSink of this.#requestSinks) { + requestSink.onOperationStatusChanged?.(result, previousStatus); + } } public onOperationHeader(operationId: string, completed: number, total: number): void { this.#workspaceSink?.onOperationHeader?.(operationId, completed, total); - this.#requestSink?.onOperationHeader?.(operationId, completed, total); + for (const requestSink of this.#requestSinks) { + requestSink.onOperationHeader?.(operationId, completed, total); + } } public onOperationChunk(operationId: string, chunk: ITerminalChunk): void { this.#workspaceSink?.onOperationChunk?.(operationId, chunk); - this.#requestSink?.onOperationChunk?.(operationId, chunk); + for (const requestSink of this.#requestSinks) { + requestSink.onOperationChunk?.(operationId, chunk); + } } public onOperationStreamClosed(operationId: string): void { this.#workspaceSink?.onOperationStreamClosed?.(operationId); - this.#requestSink?.onOperationStreamClosed?.(operationId); + for (const requestSink of this.#requestSinks) { + requestSink.onOperationStreamClosed?.(operationId); + } } public onActivity(text: string, options?: _IOperationActivityOptions): void { this.#workspaceSink?.onActivity?.(text, options); - this.#requestSink?.onActivity?.(text, options); + for (const requestSink of this.#requestSinks) { + requestSink.onActivity?.(text, options); + } } } diff --git a/libraries/rush-daemon/src/PhasedRequestEventSink.ts b/libraries/rush-daemon/src/PhasedRequestEventSink.ts index a2f137bbb0..1650bd1777 100644 --- a/libraries/rush-daemon/src/PhasedRequestEventSink.ts +++ b/libraries/rush-daemon/src/PhasedRequestEventSink.ts @@ -41,11 +41,11 @@ interface IEventOptions { class OrderedClientWriter { readonly #client: IPhasedRequestClient; - readonly #onFailure: () => void; + readonly #onFailure: (error: Error) => void; #failure: Error | undefined; #tail: Promise = Promise.resolve(); - public constructor(client: IPhasedRequestClient, onFailure: () => void) { + public constructor(client: IPhasedRequestClient, onFailure: (error: Error) => void) { this.#client = client; this.#onFailure = onFailure; } @@ -78,7 +78,7 @@ class OrderedClientWriter { await writeAsync(); } catch (error) { this.#failure = error instanceof Error ? error : new Error(String(error)); - this.#onFailure(); + this.#onFailure(this.#failure); } }); } @@ -96,7 +96,7 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink { activeOperationIds: ReadonlySet; client: IPhasedRequestClient; getNextSequence: () => number; - onWriteFailure: () => void; + onWriteFailure: (error: Error) => void; rushVersion: string; }) { this.#activeOperationIds = options.activeOperationIds; diff --git a/libraries/rush-daemon/src/PhasedRequestRouter.ts b/libraries/rush-daemon/src/PhasedRequestRouter.ts index 3304dce075..990ca2a1d8 100644 --- a/libraries/rush-daemon/src/PhasedRequestRouter.ts +++ b/libraries/rush-daemon/src/PhasedRequestRouter.ts @@ -26,12 +26,12 @@ import { import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; import { classifyRushCommand } from './RushCommandRequestPolicy'; import { + type IRequestLease, RequestExclusivityClass, RequestScheduler, RequestSchedulerError, RequestSchedulerErrorCode } from './RequestScheduler'; -import type { IRequestLease } from './RequestScheduler'; import { getRequestAdmissionErrorCode, getWorkspaceRequestScheduler, @@ -50,24 +50,47 @@ interface IDualEmitOperationGraph extends IOperationGraph { } interface IResolvedSelection { + readonly activeOperations: ReadonlyArray; readonly enabledOperations: ReadonlyArray; readonly ignoreDependencyOperations: ReadonlyArray; } interface IGraphRoutingState { - readonly graphExecutionScheduler: RequestScheduler; + readonly coordinator: PhasedRequestBatchCoordinator; readonly multiplexer: PhasedRequestEventMultiplexer; } +interface IPreparedPhasedRequest { + readonly client: IPhasedRequestClient; + readonly exclusivityClass: RequestExclusivityClass; + readonly interactiveSession: IInteractiveRequestSession | undefined; + readonly request: IDaemonPhasedRequest; + readonly selection: IResolvedSelection; + readonly warningsAllowedByEnvironment: boolean; +} + +interface IBatchEntry extends IPreparedPhasedRequest { + abortListener: (() => void) | undefined; + abortRequested: boolean; + completed: boolean; + executionStarted: boolean; + outputError: unknown; + participated: boolean; + reject: (error: unknown) => void; + requestSink: PhasedRequestEventSink | undefined; + resolve: (result: IDaemonPhasedRequestResult) => void; + unsubscribe: (() => void) | undefined; +} + const ROUTING_STATE_BY_GRAPH: WeakMap = new WeakMap(); /** * Routes one caller-resolved phased request through a real warm workspace operation graph. * * @remarks - * Command parsing, plugin loading, and graph construction remain integration-owned. Requests are serialized because - * shared-build selection merging is a later layer. Cancellation aborts only the current iteration and never closes - * the daemon-owned graph or its runners. + * Command parsing, plugin loading, and graph construction remain integration-owned. Compatible shared-build requests + * admitted before an iteration starts are merged into one graph execution. Cancellation never closes the daemon-owned + * graph or its runners. * * @beta */ @@ -98,7 +121,11 @@ export class PhasedRequestRouter { throw new DaemonRequiresInProcessError(policy); } const graph: IDualEmitOperationGraph = getDualEmitGraph(this.#workspaceSession); - const routingState: IGraphRoutingState = getGraphRoutingState(graph); + const routingState: IGraphRoutingState = getGraphRoutingState(graph, this.#workspaceSession); + const exclusivityClass: RequestExclusivityClass = classifyRushCommand({ + commandName: request.commandName, + commandOrigin: request.commandOrigin + }); let admissionController: RequestAdmissionController | undefined; let admissionLease: IRequestLease; try { @@ -109,10 +136,7 @@ export class PhasedRequestRouter { }); admissionLease = await admissionController.acquireAsync( getWorkspaceRequestScheduler(this.#workspaceSession), - classifyRushCommand({ - commandName: request.commandName, - commandOrigin: request.commandOrigin - }) + exclusivityClass ); } catch (error) { admissionController?.dispose(); @@ -120,164 +144,415 @@ export class PhasedRequestRouter { } try { - let graphLease: IRequestLease; - try { - graphLease = await admissionController.acquireAsync( - routingState.graphExecutionScheduler, - RequestExclusivityClass.Exclusive - ); - } catch (error) { - return await finishAfterAdmissionErrorAsync(request, client, interactiveSession, error); - } let inputAttachment: Disposable | undefined; try { inputAttachment = attachInteractiveInput(request, client, interactiveSession); try { - return await this.#executeAdmittedAsync( - request, - client, - graph, - routingState, - interactiveSession + validateEngineShape(request.engineShape, this.#workspaceSession.engineShape); + const operationById: ReadonlyMap = indexOperations(graph.operations); + const selection: IResolvedSelection = resolveSelection(request.operationSelection, operationById); + let warningsAllowedByEnvironment: boolean; + try { + warningsAllowedByEnvironment = parseWarningsAllowedByEnvironment(request.environment); + } catch (error) { + const cleanupErrors: unknown[] = []; + await collectInteractiveCleanupErrorAsync(interactiveSession, cleanupErrors); + const result: IDaemonPhasedRequestResult = createPhasedCommandResult({ + aborted: client.abortSignal.aborted, + error: combineErrors(error, cleanupErrors), + graphStatus: graph.status, + operationOutcomes: [], + requestId: request.requestId, + scheduled: false, + warningsAllowedByEnvironment: false + }); + await client.writeResultAsync(result); + return result; + } + if (client.abortSignal.aborted) { + return await writeAbortedResultAsync(request.requestId, client, interactiveSession); + } + return await routingState.coordinator.enqueueAsync( + { + client, + exclusivityClass, + interactiveSession, + request, + selection, + warningsAllowedByEnvironment + }, + admissionController ); } catch (error) { + if (error instanceof RequestSchedulerError) { + return await finishAfterAdmissionErrorAsync(request, client, interactiveSession, error); + } return await finishAfterRoutingErrorAsync(interactiveSession, error); } } finally { inputAttachment?.[Symbol.dispose](); - graphLease.release(); } } finally { admissionLease.release(); admissionController.dispose(); } } +} - async #executeAdmittedAsync( - request: IDaemonPhasedRequest, - client: IPhasedRequestClient, +class PhasedRequestBatchCoordinator { + readonly #graph: IDualEmitOperationGraph; + readonly #graphExecutionScheduler: RequestScheduler; + readonly #multiplexer: PhasedRequestEventMultiplexer; + readonly #pending: IBatchEntry[] = []; + readonly #workspaceSession: IWorkspaceSession; + readonly #abortErrors: unknown[] = []; + #abortTail: Promise = Promise.resolve(); + #acceptingCurrentBatch: boolean = false; + #currentBatch: ReadonlyArray | undefined; + #drainScheduled: boolean = false; + #running: boolean = false; + + public constructor( graph: IDualEmitOperationGraph, - routingState: IGraphRoutingState, - interactiveSession: IInteractiveRequestSession | undefined + graphExecutionScheduler: RequestScheduler, + multiplexer: PhasedRequestEventMultiplexer, + workspaceSession: IWorkspaceSession + ) { + this.#graph = graph; + this.#graphExecutionScheduler = graphExecutionScheduler; + this.#multiplexer = multiplexer; + this.#workspaceSession = workspaceSession; + } + + public async enqueueAsync( + request: IPreparedPhasedRequest, + admissionController: RequestAdmissionController ): Promise { - validateEngineShape(request.engineShape, this.#workspaceSession.engineShape); - const operationById: ReadonlyMap = indexOperations(graph.operations); - const selection: IResolvedSelection = resolveSelection(request.operationSelection, operationById); - let warningsAllowedByEnvironment: boolean; - try { - warningsAllowedByEnvironment = parseWarningsAllowedByEnvironment(request.environment); - } catch (error) { - const cleanupErrors: unknown[] = []; - await collectInteractiveCleanupErrorAsync(interactiveSession, cleanupErrors); - const result: IDaemonPhasedRequestResult = createPhasedCommandResult({ - aborted: client.abortSignal.aborted, - error: combineErrors(error, cleanupErrors), - graphStatus: graph.status, - operationOutcomes: [], - requestId: request.requestId, - scheduled: false, - warningsAllowedByEnvironment: false - }); - await client.writeResultAsync(result); - return result; + if (!this.#canJoinCurrentBatch(request)) { + const graphExclusivityClass: RequestExclusivityClass = + request.exclusivityClass === RequestExclusivityClass.SharedBuild + ? RequestExclusivityClass.SharedBuild + : RequestExclusivityClass.Exclusive; + const graphWaitLease: IRequestLease = await admissionController.acquireAsync( + this.#graphExecutionScheduler, + graphExclusivityClass + ); + graphWaitLease.release(); } + return new Promise((resolve, reject) => { + const entry: IBatchEntry = { + ...request, + abortListener: undefined, + abortRequested: false, + completed: false, + executionStarted: false, + outputError: undefined, + participated: false, + reject, + requestSink: undefined, + resolve, + unsubscribe: undefined + }; + entry.abortListener = () => this.#deactivateEntry(entry, true); + request.client.abortSignal.addEventListener('abort', entry.abortListener, { once: true }); + this.#pending.push(entry); + this.#scheduleDrain(); + }); + } - if (client.abortSignal.aborted) { - return await writeAbortedResultAsync(request.requestId, client, interactiveSession); + #scheduleDrain(): void { + if (this.#running || this.#drainScheduled) { + return; } - if (graph.hasScheduledIteration || graph.status === OperationStatus.Executing) { - throw new Error('The warm workspace operation graph is not idle.'); + this.#drainScheduled = true; + setImmediate(() => { + this.#drainScheduled = false; + void this.#drainAsync(); + }); + } + + async #drainAsync(): Promise { + if (this.#running) { + return; } - await this.#workspaceSession.reconcileInvalidationsAsync(); - if (client.abortSignal.aborted) { - return await writeAbortedResultAsync(request.requestId, client, interactiveSession); + this.#running = true; + try { + while (this.#pending.length > 0) { + const first: IBatchEntry = this.#pending.shift()!; + const batch: IBatchEntry[] = [first]; + if (first.exclusivityClass === RequestExclusivityClass.SharedBuild) { + this.#takeCompatiblePending(batch); + } + this.#currentBatch = batch; + this.#acceptingCurrentBatch = + first.exclusivityClass === RequestExclusivityClass.SharedBuild; + for (const entry of batch) { + entry.executionStarted = true; + } + try { + await this.#executeBatchAsync(batch); + } catch (error) { + await Promise.all(batch.map((entry: IBatchEntry) => this.#rejectEntryAsync(entry, error))); + } finally { + this.#acceptingCurrentBatch = false; + this.#currentBatch = undefined; + } + } + } finally { + this.#running = false; + if (this.#pending.length > 0) { + this.#scheduleDrain(); + } } + } - applySelection(graph, selection); - const activeOperations: ReadonlyArray = Array.from(graph.operations).filter( - (operation: Operation) => operation.enabled !== false - ); - const activeOperationIds: ReadonlySet = new Set( - activeOperations.map((operation: Operation) => operation.name) + #canJoinCurrentBatch(request: IPreparedPhasedRequest): boolean { + if (!this.#running) { + return true; + } + return ( + this.#acceptingCurrentBatch && + request.exclusivityClass === RequestExclusivityClass.SharedBuild && + this.#currentBatch?.[0]?.exclusivityClass === RequestExclusivityClass.SharedBuild ); + } - let abortTail: Promise = Promise.resolve(); - const abortErrors: unknown[] = []; - let wasAborted: boolean = false; - const abortIteration = (): void => { - wasAborted = true; - abortTail = abortTail - .then(() => graph.abortCurrentIterationAsync()) - .catch((error: unknown) => { - abortErrors.push(error); - }); - }; - const previousPauseNextIteration: boolean = graph.pauseNextIteration; - const requestSink: PhasedRequestEventSink = new PhasedRequestEventSink({ - activeOperationIds, - client, - getNextSequence: () => client.getNextEventSequence(), - onWriteFailure: abortIteration, - rushVersion: this.#workspaceSession.metadata.rushVersion - }); - const unsubscribe: () => void = routingState.multiplexer.subscribe(requestSink); - setPauseNextIteration(graph, true); - client.abortSignal.addEventListener('abort', abortIteration, { once: true }); + #takeCompatiblePending(batch: IBatchEntry[]): void { + for (let index: number = 0; index < this.#pending.length; ) { + const entry: IBatchEntry = this.#pending[index]; + if (entry.exclusivityClass === RequestExclusivityClass.SharedBuild) { + this.#pending.splice(index, 1); + entry.executionStarted = true; + batch.push(entry); + } else { + index++; + } + } + } - let scheduled: boolean = false; - let executionError: unknown; - const iterationCleanupErrors: unknown[] = []; + async #executeBatchAsync(batch: IBatchEntry[]): Promise { + const graphLeasePromise: Promise = this.#graphExecutionScheduler.acquireAsync({ + exclusivityClass: RequestExclusivityClass.Exclusive + }); + const graphLease: IRequestLease = await graphLeasePromise; try { - scheduled = await graph.scheduleIterationAsync({ - inputsSnapshot: this.#workspaceSession.inputsSnapshot - }); - if (scheduled) { - const executionPromise: Promise = graph.executeScheduledIterationAsync(); - if (wasAborted || client.abortSignal.aborted) { - await Promise.resolve(); - abortIteration(); - } - await executionPromise; + if (this.#graph.hasScheduledIteration || this.#graph.status === OperationStatus.Executing) { + throw new Error('The warm workspace operation graph is not idle.'); } - } catch (error) { - executionError = error; - if (graph.hasScheduledIteration) { - try { - const failedExecutionPromise: Promise = graph.executeScheduledIterationAsync(); - await Promise.resolve(); - abortIteration(); - await failedExecutionPromise; - } catch (cleanupError) { - iterationCleanupErrors.push(cleanupError); + await this.#workspaceSession.reconcileInvalidationsAsync(); + + if (batch[0].exclusivityClass === RequestExclusivityClass.SharedBuild) { + this.#takeCompatiblePending(batch); + } + this.#acceptingCurrentBatch = false; + const participants: IBatchEntry[] = batch.filter((entry: IBatchEntry) => + this.#isEntryLive(entry) + ); + if (participants.length === 0) { + await Promise.all( + batch.map((entry: IBatchEntry) => this.#finishEntryAsync(entry, false, undefined)) + ); + return; + } + + applySelections( + this.#graph, + participants.map((entry: IBatchEntry) => entry.selection) + ); + for (const entry of participants) { + entry.participated = true; + const activeOperationIds: ReadonlySet = new Set( + entry.selection.activeOperations.map((operation: Operation) => operation.name) + ); + entry.requestSink = new PhasedRequestEventSink({ + activeOperationIds, + client: entry.client, + getNextSequence: () => entry.client.getNextEventSequence(), + onWriteFailure: (error: Error) => this.#deactivateEntry(entry, false, error), + rushVersion: this.#workspaceSession.metadata.rushVersion + }); + entry.unsubscribe = this.#multiplexer.subscribe(entry.requestSink); + } + + const previousPauseNextIteration: boolean = this.#graph.pauseNextIteration; + setPauseNextIteration(this.#graph, true); + let scheduled: boolean = false; + let executionError: unknown; + const iterationCleanupErrors: unknown[] = []; + try { + scheduled = await this.#graph.scheduleIterationAsync({ + inputsSnapshot: this.#workspaceSession.inputsSnapshot + }); + if (scheduled) { + await Promise.all( + participants.map(async (entry: IBatchEntry) => { + try { + await entry.requestSink?.flushAsync(); + } catch { + // The sink already recorded the write error and deactivated this client. + } + }) + ); + const executionPromise: Promise = this.#graph.executeScheduledIterationAsync(); + if (!participants.some((entry: IBatchEntry) => this.#isEntryLive(entry))) { + // Let executeScheduledIterationAsync promote the scheduled iteration before aborting it. + await Promise.resolve(); + this.#requestIterationAbort(); + await this.#abortTail; + } + await executionPromise; } + } catch (error) { + executionError = error; + if (this.#graph.hasScheduledIteration) { + try { + const failedExecutionPromise: Promise = + this.#graph.executeScheduledIterationAsync(); + await Promise.resolve(); + this.#requestIterationAbort(); + await this.#abortTail; + await failedExecutionPromise; + } catch (cleanupError) { + iterationCleanupErrors.push(cleanupError); + } + } + } finally { + for (const entry of participants) { + entry.unsubscribe?.(); + entry.unsubscribe = undefined; + } + setPauseNextIteration(this.#graph, previousPauseNextIteration); + } + + await this.#abortTail; + iterationCleanupErrors.push(...this.#abortErrors.splice(0)); + await Promise.all( + batch.map((entry: IBatchEntry) => + this.#finishEntryAsync(entry, scheduled, executionError, iterationCleanupErrors) + ) + ); + } finally { + graphLease.release(); + } + } + + #deactivateEntry(entry: IBatchEntry, aborted: boolean, outputError?: Error): void { + if (entry.completed) { + return; + } + if (aborted) { + entry.abortRequested = true; + } else { + entry.outputError ??= outputError ?? new Error('The phased request client output failed.'); + } + entry.unsubscribe?.(); + entry.unsubscribe = undefined; + + if (!entry.executionStarted) { + const pendingIndex: number = this.#pending.indexOf(entry); + if (pendingIndex >= 0) { + this.#pending.splice(pendingIndex, 1); + void this.#finishEntryAsync(entry, false, undefined).catch((error: unknown) => { + this.#completeEntry(entry); + entry.reject(error); + }); + return; } } - await abortTail; - unsubscribe(); - setPauseNextIteration(graph, previousPauseNextIteration); - client.abortSignal.removeEventListener('abort', abortIteration); - const cleanupErrors: unknown[] = [...iterationCleanupErrors, ...abortErrors]; - const observedAbortErrorCount: number = abortErrors.length; - try { - await requestSink.flushAsync(); - } catch (error) { - cleanupErrors.push(error); + if ( + entry.executionStarted && + this.#currentBatch && + (this.#graph.hasScheduledIteration || this.#graph.status === OperationStatus.Executing) && + !this.#currentBatch.some((candidate: IBatchEntry) => this.#isEntryLive(candidate)) + ) { + this.#requestIterationAbort(); } - await collectInteractiveCleanupErrorAsync(interactiveSession, cleanupErrors); - await abortTail; - cleanupErrors.push(...abortErrors.slice(observedAbortErrorCount)); + } + + #isEntryLive(entry: IBatchEntry): boolean { + return !entry.abortRequested && !entry.client.abortSignal.aborted && entry.outputError === undefined; + } + + #requestIterationAbort(): void { + const abortPromise: Promise = this.#graph.abortCurrentIterationAsync(); + this.#abortTail = Promise.all([this.#abortTail, abortPromise]) + .then(() => undefined) + .catch((error: unknown) => { + this.#abortErrors.push(error); + }); + } + + async #finishEntryAsync( + entry: IBatchEntry, + batchScheduled: boolean, + executionError: unknown, + batchCleanupErrors: ReadonlyArray = [] + ): Promise { + if (entry.completed) { + return; + } + const cleanupErrors: unknown[] = [...batchCleanupErrors]; + if (entry.requestSink) { + try { + await entry.requestSink.flushAsync(); + } catch (error) { + cleanupErrors.push(error); + } + } + await collectInteractiveCleanupErrorAsync(entry.interactiveSession, cleanupErrors); + const aborted: boolean = entry.abortRequested || entry.client.abortSignal.aborted; + const operationOutcomes: ReadonlyArray = entry.requestSink + ? collectOperationOutcomes( + entry.selection.activeOperations, + this.#graph, + entry.requestSink, + aborted && entry.participated + ) + : []; const result: IDaemonPhasedRequestResult = createPhasedCommandResult({ - aborted: wasAborted || client.abortSignal.aborted, + aborted, error: combineErrors(executionError, cleanupErrors), - graphStatus: graph.status, - operationOutcomes: collectOperationOutcomes(activeOperations, graph, requestSink), - requestId: request.requestId, - scheduled, - warningsAllowedByEnvironment + graphStatus: getClientGraphStatus(aborted, operationOutcomes), + operationOutcomes, + requestId: entry.request.requestId, + scheduled: entry.participated && batchScheduled, + warningsAllowedByEnvironment: entry.warningsAllowedByEnvironment }); - await client.writeResultAsync(result); - return result; + try { + await entry.client.writeResultAsync(result); + this.#completeEntry(entry); + entry.resolve(result); + } catch (error) { + this.#completeEntry(entry); + entry.reject(error); + } + } + + async #rejectEntryAsync(entry: IBatchEntry, error: unknown): Promise { + if (entry.completed) { + return; + } + entry.unsubscribe?.(); + entry.unsubscribe = undefined; + const cleanupErrors: unknown[] = []; + if (entry.requestSink) { + try { + await entry.requestSink.flushAsync(); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + } + this.#completeEntry(entry); + entry.reject(combineErrors(error, cleanupErrors)); + } + + #completeEntry(entry: IBatchEntry): void { + entry.completed = true; + if (entry.abortListener) { + entry.client.abortSignal.removeEventListener('abort', entry.abortListener); + entry.abortListener = undefined; + } } } @@ -307,14 +582,23 @@ function setPauseNextIteration(graph: IOperationGraph, pauseNextIteration: boole graph.pauseNextIteration = pauseNextIteration; } -function getGraphRoutingState(graph: IDualEmitOperationGraph): IGraphRoutingState { +function getGraphRoutingState( + graph: IDualEmitOperationGraph, + workspaceSession: IWorkspaceSession +): IGraphRoutingState { let state: IGraphRoutingState | undefined = ROUTING_STATE_BY_GRAPH.get(graph); if (!state) { const multiplexer: PhasedRequestEventMultiplexer = new PhasedRequestEventMultiplexer( getGraphEventSink(graph) ); + const graphExecutionScheduler: RequestScheduler = new RequestScheduler(); state = { - graphExecutionScheduler: new RequestScheduler(), + coordinator: new PhasedRequestBatchCoordinator( + graph, + graphExecutionScheduler, + multiplexer, + workspaceSession + ), multiplexer }; ROUTING_STATE_BY_GRAPH.set(graph, state); @@ -417,7 +701,11 @@ function resolveSelection( } addSelectedOperation(selection.enabledState, operation, enabledOperations, ignoreDependencyOperations); } - return { enabledOperations, ignoreDependencyOperations }; + return { + activeOperations: collectSelectionClosure(enabledOperations, ignoreDependencyOperations), + enabledOperations, + ignoreDependencyOperations + }; } function addSelectedOperation( @@ -435,16 +723,43 @@ function addSelectedOperation( } } -function applySelection(graph: IOperationGraph, selection: IResolvedSelection): void { +function collectSelectionClosure( + enabledOperations: ReadonlyArray, + ignoreDependencyOperations: ReadonlyArray +): ReadonlyArray { + const activeOperations: Set = new Set([ + ...enabledOperations, + ...ignoreDependencyOperations + ]); + for (const operation of activeOperations) { + for (const dependency of operation.dependencies) { + activeOperations.add(dependency); + } + } + return Array.from(activeOperations); +} + +function applySelections( + graph: IOperationGraph, + selections: ReadonlyArray +): void { graph.setEnabledStates(graph.operations, false, 'unsafe'); graph.setEnabledStates( - selection.ignoreDependencyOperations, + selections.flatMap( + (selection: IResolvedSelection) => selection.ignoreDependencyOperations + ), 'ignore-dependency-changes', 'safe' ); - graph.setEnabledStates(selection.enabledOperations, true, 'safe'); graph.setEnabledStates( - selection.ignoreDependencyOperations, + selections.flatMap((selection: IResolvedSelection) => selection.enabledOperations), + true, + 'safe' + ); + graph.setEnabledStates( + selections.flatMap( + (selection: IResolvedSelection) => selection.ignoreDependencyOperations + ), 'ignore-dependency-changes', 'unsafe' ); @@ -453,14 +768,18 @@ function applySelection(graph: IOperationGraph, selection: IResolvedSelection): function collectOperationOutcomes( activeOperations: ReadonlyArray, graph: IOperationGraph, - requestSink: PhasedRequestEventSink + requestSink: PhasedRequestEventSink, + fillMissingAsAborted: boolean = false ): ReadonlyArray { const outcomes: IPhasedOperationOutcome[] = []; for (const operation of [...activeOperations].sort(compareOperations)) { const observed: ReturnType = requestSink.getObservedResult(operation); const retained: IOperationExecutionResult | undefined = graph.resultByOperation.get(operation); - const status: string | undefined = observed?.status ?? retained?.status; + const status: string | undefined = + retained?.status ?? + observed?.status ?? + (fillMissingAsAborted ? OperationStatus.Aborted : undefined); if (status === undefined) { continue; } @@ -480,6 +799,38 @@ function compareOperations(left: Operation, right: Operation): number { return left.name.localeCompare(right.name); } +function getClientGraphStatus( + aborted: boolean, + operationOutcomes: ReadonlyArray +): OperationStatus { + if ( + operationOutcomes.some( + ({ result }: IPhasedOperationOutcome) => + result.status === OperationStatus.Failure || result.status === OperationStatus.Blocked + ) + ) { + return OperationStatus.Failure; + } + if (aborted) { + return OperationStatus.Aborted; + } + if ( + operationOutcomes.some( + ({ result }: IPhasedOperationOutcome) => result.status === OperationStatus.Aborted + ) + ) { + return OperationStatus.Aborted; + } + if ( + operationOutcomes.some( + ({ result }: IPhasedOperationOutcome) => result.status === OperationStatus.SuccessWithWarning + ) + ) { + return OperationStatus.SuccessWithWarning; + } + return OperationStatus.Success; +} + async function writeAbortedResultAsync( requestId: string, client: IPhasedRequestClient, diff --git a/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts b/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts new file mode 100644 index 0000000000..a21d17ca13 --- /dev/null +++ b/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts @@ -0,0 +1,443 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { ITerminal } from '@rushstack/terminal'; +import type { + IDaemonEventEnvelope, + IDaemonPhasedOperationSelection, + IDaemonPhasedRequest, + IDaemonPhasedRequestResult +} from '@rushstack/rush-daemon-protocol'; +import { OperationStatus } from '@microsoft/rush-lib'; + +import { PhasedRequestRouter } from '../PhasedRequestRouter'; +import { + TEST_ENGINE_SHAPE, + TestOperationRunner, + TestPhasedRequestClient, + createRoutingFixture +} from './PhasedRequestRouterTestUtilities'; +import type { ITestRoutingFixture } from './PhasedRequestRouterTestUtilities'; + +const OPERATION_A: string = 'project-a (_phase:test)'; +const OPERATION_B: string = 'project-b (_phase:test)'; +const OPERATION_C: string = 'project-c (_phase:test)'; + +interface IDeferred { + readonly promise: Promise; + readonly resolve: () => void; +} + +function createDeferred(): IDeferred { + let resolvePromise: (() => void) | undefined; + const promise: Promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + return { promise, resolve: () => resolvePromise?.() }; +} + +function select(operationId: string): IDaemonPhasedOperationSelection { + return { enabledState: true, operationId }; +} + +function createRequest( + requestId: string, + ...selectedOperationIds: ReadonlyArray +): IDaemonPhasedRequest { + return { + commandName: 'build', + commandOrigin: 'built-in', + engineShape: TEST_ENGINE_SHAPE, + environment: {}, + operationSelection: selectedOperationIds.map(select), + requestId + }; +} + +function createFixture(options?: { + readonly actionAAsync?: (terminal: ITerminal) => Promise; + readonly actionCAsync?: (terminal: ITerminal) => Promise; + readonly statusA?: OperationStatus; +}): ITestRoutingFixture { + return createRoutingFixture( + new Map([ + [ + OPERATION_A, + new TestOperationRunner( + OPERATION_A, + options?.statusA ?? OperationStatus.Success, + options?.actionAAsync + ) + ], + [OPERATION_B, new TestOperationRunner(OPERATION_B)], + [OPERATION_C, new TestOperationRunner(OPERATION_C, OperationStatus.Success, options?.actionCAsync)] + ]), + [[OPERATION_B, OPERATION_A]] + ); +} + +function getResultOperationIds(result: IDaemonPhasedRequestResult): ReadonlyArray { + return result.operationResults.map(({ operationId }) => operationId); +} + +function eventOperationId(event: IDaemonEventEnvelope): string | undefined { + if (event.scope?.operationId) { + return event.scope.operationId; + } + const payload: unknown = event.payload; + if (typeof payload !== 'object' || payload === null) { + return undefined; + } + const operationId: unknown = (payload as { operationId?: unknown }).operationId; + if (typeof operationId === 'string') { + return operationId; + } + const data: unknown = (payload as { data?: unknown }).data; + return typeof data === 'object' && data !== null + ? ((data as { operationId?: string }).operationId ?? undefined) + : undefined; +} + +describe('shared phased request batching', () => { + it('merges overlapping selections into one real graph iteration and executes shared operations once', async () => { + const fixture: ITestRoutingFixture = createFixture(); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + + const [dependency, consumer] = await Promise.all([ + router.executeAsync(createRequest('dependency', OPERATION_A), new TestPhasedRequestClient('one')), + router.executeAsync(createRequest('consumer', OPERATION_B), new TestPhasedRequestClient('two')) + ]); + + expect(scheduleSpy).toHaveBeenCalledTimes(1); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(1); + expect(fixture.runners.get(OPERATION_B)?.runCount).toBe(1); + expect(getResultOperationIds(dependency)).toEqual([OPERATION_A]); + expect(getResultOperationIds(consumer)).toEqual([OPERATION_A, OPERATION_B]); + }); + + it('shares one iteration for disjoint selections while isolating streams, events, and results', async () => { + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (terminal: ITerminal): Promise => terminal.writeLine('only-a'), + actionCAsync: async (terminal: ITerminal): Promise => terminal.writeLine('only-c') + }); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + const clientA: TestPhasedRequestClient = new TestPhasedRequestClient('one'); + const clientC: TestPhasedRequestClient = new TestPhasedRequestClient('two'); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + + const [resultA, resultC] = await Promise.all([ + router.executeAsync(createRequest('a', OPERATION_A), clientA), + router.executeAsync(createRequest('c', OPERATION_C), clientC) + ]); + + expect(scheduleSpy).toHaveBeenCalledTimes(1); + expect(getResultOperationIds(resultA)).toEqual([OPERATION_A]); + expect(getResultOperationIds(resultC)).toEqual([OPERATION_C]); + expect(getWrittenOperationIds(clientA)).toEqual(new Set([OPERATION_A])); + expect(getWrittenOperationIds(clientC)).toEqual(new Set([OPERATION_C])); + }); + + it('derives shared and disjoint failure results from each client subset', async () => { + const fixture: ITestRoutingFixture = createFixture({ statusA: OperationStatus.Failure }); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + + const [failed, blocked, disjoint] = await Promise.all([ + router.executeAsync(createRequest('failed', OPERATION_A), new TestPhasedRequestClient('one')), + router.executeAsync(createRequest('blocked', OPERATION_B), new TestPhasedRequestClient('two')), + router.executeAsync(createRequest('disjoint', OPERATION_C), new TestPhasedRequestClient('three')) + ]); + + expect(failed).toMatchObject({ exitCode: 1, outcome: 'failure' }); + expect(blocked).toMatchObject({ exitCode: 1, outcome: 'failure' }); + expect(blocked.operationResults).toEqual( + expect.arrayContaining([ + expect.objectContaining({ operationId: OPERATION_A, status: OperationStatus.Failure }), + expect.objectContaining({ operationId: OPERATION_B, status: OperationStatus.Blocked }) + ]) + ); + expect(disjoint).toMatchObject({ exitCode: 0, outcome: 'success' }); + expect(getResultOperationIds(disjoint)).toEqual([OPERATION_C]); + }); + + it('removes a client cancelled before scheduling without running its selection', async () => { + const fixture: ITestRoutingFixture = createFixture(); + const reconcileStarted: IDeferred = createDeferred(); + const releaseReconcile: IDeferred = createDeferred(); + fixture.session.onReconcileAsync = async (): Promise => { + reconcileStarted.resolve(); + await releaseReconcile.promise; + }; + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + const resultPromise: Promise = new PhasedRequestRouter( + fixture.session + ).executeAsync(createRequest('cancelled', OPERATION_A), client); + await reconcileStarted.promise; + + client.abortController.abort(); + releaseReconcile.resolve(); + const result: IDaemonPhasedRequestResult = await resultPromise; + + expect(result).toMatchObject({ aborted: true, outcome: 'aborted', scheduled: false }); + expect(scheduleSpy).not.toHaveBeenCalled(); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(0); + }); + + it('unsubscribes one mid-run cancellation without aborting work required by another client', async () => { + const operationStarted: IDeferred = createDeferred(); + const releaseOperation: IDeferred = createDeferred(); + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (): Promise => { + operationStarted.resolve(); + await releaseOperation.promise; + } + }); + const cancelledClient: TestPhasedRequestClient = new TestPhasedRequestClient('one'); + const continuingClient: TestPhasedRequestClient = new TestPhasedRequestClient('two'); + const abortSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'abortCurrentIterationAsync'); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const cancelled = router.executeAsync(createRequest('cancelled', OPERATION_A), cancelledClient); + const continuing = router.executeAsync(createRequest('continuing', OPERATION_C), continuingClient); + await operationStarted.promise; + const abortCallCountBeforeCancellation: number = abortSpy.mock.calls.length; + + cancelledClient.abortController.abort(); + releaseOperation.resolve(); + const [cancelledResult, continuingResult] = await Promise.all([cancelled, continuing]); + + expect(cancelledResult).toMatchObject({ aborted: true, outcome: 'aborted' }); + expect(continuingResult).toMatchObject({ exitCode: 0, outcome: 'success' }); + expect(abortSpy).toHaveBeenCalledTimes(abortCallCountBeforeCancellation); + expect(fixture.runners.get(OPERATION_C)?.runCount).toBe(1); + }); + + it('reports authoritative retained status when a client cancels during a shared operation', async () => { + const operationStarted: IDeferred = createDeferred(); + const releaseOperation: IDeferred = createDeferred(); + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (): Promise => { + operationStarted.resolve(); + await releaseOperation.promise; + } + }); + const cancelledClient: TestPhasedRequestClient = new TestPhasedRequestClient('one'); + const continuingClient: TestPhasedRequestClient = new TestPhasedRequestClient('two'); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const cancelled = router.executeAsync(createRequest('cancelled', OPERATION_A), cancelledClient); + const continuing = router.executeAsync(createRequest('continuing', OPERATION_A), continuingClient); + await operationStarted.promise; + + cancelledClient.abortController.abort(); + releaseOperation.resolve(); + const [cancelledResult, continuingResult] = await Promise.all([cancelled, continuing]); + + expect(cancelledResult).toMatchObject({ aborted: true, outcome: 'aborted' }); + expect(cancelledResult.operationResults).toEqual([ + expect.objectContaining({ operationId: OPERATION_A, status: OperationStatus.Success }) + ]); + expect(continuingResult).toMatchObject({ exitCode: 0, outcome: 'success' }); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(1); + }); + + it('preserves failure precedence when a client cancels during a failing shared operation', async () => { + const operationStarted: IDeferred = createDeferred(); + const releaseOperation: IDeferred = createDeferred(); + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (): Promise => { + operationStarted.resolve(); + await releaseOperation.promise; + }, + statusA: OperationStatus.Failure + }); + const cancelledClient: TestPhasedRequestClient = new TestPhasedRequestClient('one'); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const cancelled = router.executeAsync(createRequest('cancelled', OPERATION_A), cancelledClient); + const continuing = router.executeAsync( + createRequest('continuing', OPERATION_A), + new TestPhasedRequestClient('two') + ); + await operationStarted.promise; + + cancelledClient.abortController.abort(); + releaseOperation.resolve(); + const [cancelledResult, continuingResult] = await Promise.all([cancelled, continuing]); + + expect(cancelledResult).toMatchObject({ aborted: true, exitCode: 1, outcome: 'failure' }); + expect(cancelledResult.operationResults).toEqual([ + expect.objectContaining({ operationId: OPERATION_A, status: OperationStatus.Failure }) + ]); + expect(continuingResult).toMatchObject({ aborted: false, exitCode: 1, outcome: 'failure' }); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(1); + }); + + it('aborts the shared iteration when every client cancels', async () => { + const operationStarted: IDeferred = createDeferred(); + const releaseOperation: IDeferred = createDeferred(); + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (): Promise => { + operationStarted.resolve(); + await releaseOperation.promise; + } + }); + const firstClient: TestPhasedRequestClient = new TestPhasedRequestClient('one'); + const secondClient: TestPhasedRequestClient = new TestPhasedRequestClient('two'); + const abortSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'abortCurrentIterationAsync'); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const first = router.executeAsync(createRequest('first', OPERATION_A), firstClient); + const second = router.executeAsync(createRequest('second', OPERATION_B), secondClient); + await operationStarted.promise; + const abortCallCountBeforeCancellation: number = abortSpy.mock.calls.length; + + firstClient.abortController.abort(); + secondClient.abortController.abort(); + releaseOperation.resolve(); + const results: ReadonlyArray = await Promise.all([first, second]); + + expect(results).toEqual([ + expect.objectContaining({ aborted: true, outcome: 'aborted' }), + expect.objectContaining({ aborted: true, outcome: 'aborted' }) + ]); + expect(abortSpy.mock.calls.length).toBeGreaterThan(abortCallCountBeforeCancellation); + expect(fixture.runners.get(OPERATION_B)?.runCount).toBe(0); + }); + + it('puts arrivals after execution begins into a later batch and reconciles once per batch', async () => { + const operationStarted: IDeferred = createDeferred(); + const releaseOperation: IDeferred = createDeferred(); + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (): Promise => { + operationStarted.resolve(); + await releaseOperation.promise; + } + }); + fixture.session.onReconcileAsync = jest.fn(async (): Promise => undefined); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const first = router.executeAsync( + createRequest('first', OPERATION_A), + new TestPhasedRequestClient('one') + ); + await operationStarted.promise; + const late = router.executeAsync( + createRequest('late', OPERATION_C), + new TestPhasedRequestClient('two') + ); + expect(fixture.runners.get(OPERATION_C)?.runCount).toBe(0); + releaseOperation.resolve(); + + await Promise.all([first, late]); + expect(scheduleSpy).toHaveBeenCalledTimes(2); + expect(fixture.session.onReconcileAsync).toHaveBeenCalledTimes(2); + expect(fixture.runners.get(OPERATION_C)?.runCount).toBe(1); + }); + + it('serializes concurrent shared-read requests instead of merging or deadlocking them', async () => { + const operationStarted: IDeferred = createDeferred(); + const releaseOperation: IDeferred = createDeferred(); + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (): Promise => { + operationStarted.resolve(); + await releaseOperation.promise; + } + }); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const firstRequest: IDaemonPhasedRequest = { + ...createRequest('first', OPERATION_A), + commandName: 'list' + }; + const first = router.executeAsync(firstRequest, new TestPhasedRequestClient('one')); + await operationStarted.promise; + + const second = router.executeAsync( + { ...createRequest('second', OPERATION_C), commandName: 'list' }, + new TestPhasedRequestClient('two') + ); + const third = router.executeAsync( + { ...createRequest('third', OPERATION_C), commandName: 'list' }, + new TestPhasedRequestClient('three') + ); + releaseOperation.resolve(); + + await Promise.all([first, second, third]); + expect(scheduleSpy).toHaveBeenCalledTimes(3); + }); + + it('preserves per-client backpressure and final-result ordering in a merged batch', async () => { + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (terminal: ITerminal): Promise => { + terminal.writeLine('first'); + terminal.writeErrorLine('second'); + } + }); + const clients: ReadonlyArray = [ + new TestPhasedRequestClient('one'), + new TestPhasedRequestClient('two') + ]; + const concurrentWrites: number[] = [0, 0]; + const maximumConcurrentWrites: number[] = [0, 0]; + clients.forEach((client: TestPhasedRequestClient, index: number) => { + client.onWriteAsync = async (): Promise => { + concurrentWrites[index]++; + maximumConcurrentWrites[index] = Math.max( + maximumConcurrentWrites[index], + concurrentWrites[index] + ); + await new Promise((resolve) => setImmediate(resolve)); + concurrentWrites[index]--; + }; + }); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + + await Promise.all([ + router.executeAsync(createRequest('one', OPERATION_A), clients[0]), + router.executeAsync(createRequest('two', OPERATION_A), clients[1]) + ]); + + expect(maximumConcurrentWrites).toEqual([1, 1]); + for (const client of clients) { + expect(client.writes[client.writes.length - 1]?.result).toBeDefined(); + } + }); + + it('cleans up a failed batch so a later batch can execute', async () => { + const fixture: ITestRoutingFixture = createFixture(); + let schedulingCount: number = 0; + fixture.graph.hooks.onIterationScheduled.tap('fail first batch', () => { + if (schedulingCount++ === 0) { + throw new Error('first batch failed'); + } + }); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + + const first = await router.executeAsync( + createRequest('first', OPERATION_A), + new TestPhasedRequestClient('one') + ); + const second = await router.executeAsync( + createRequest('second', OPERATION_C), + new TestPhasedRequestClient('two') + ); + + expect(first).toMatchObject({ errorMessage: 'first batch failed', outcome: 'failure' }); + expect(second).toMatchObject({ exitCode: 0, outcome: 'success' }); + expect(fixture.runners.get(OPERATION_C)?.runCount).toBe(1); + }); +}); + +function getWrittenOperationIds(client: TestPhasedRequestClient): ReadonlySet { + const operationIdSet: Set = new Set(); + for (const write of client.writes) { + if (write.operationId) { + operationIdSet.add(write.operationId); + } + if (write.event) { + const operationId: string | undefined = eventOperationId(write.event); + if (operationId) { + operationIdSet.add(operationId); + } + } + } + return operationIdSet; +} diff --git a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts index f6308b7984..8b500de4a3 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts @@ -62,7 +62,7 @@ export interface ITestClientWrite { export class TestPhasedRequestClient implements IPhasedRequestClient { public readonly abortController: AbortController = new AbortController(); - public readonly sessionId: string = 'test-session'; + public readonly sessionId: string; public readonly supportsRequestAdmission: boolean = true; public readonly writes: ITestClientWrite[] = []; public readonly policies: IDaemonTerminalPolicyResult[] = []; @@ -75,6 +75,10 @@ export class TestPhasedRequestClient implements IPhasedRequestClient { this.#sequenceState = sequenceState; } + public constructor(sessionId: string = 'test-session') { + this.sessionId = sessionId; + } + public get abortSignal(): AbortSignal { return this.abortController.signal; } diff --git a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts index 68f6a0f045..382f0139e5 100644 --- a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts +++ b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts @@ -11,6 +11,7 @@ import type { IDaemonRequestQueuePositionMessage, IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; +import { OperationStatus } from '@microsoft/rush-lib'; import type { IGlobalCommandExecutionContext } from '../GlobalCommandExecutionContext'; import type { IResolvedGlobalCommandRequest } from '../GlobalCommandRequest'; @@ -349,7 +350,7 @@ describe('request admission integration', () => { expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(1); }); - it('applies no-wait and one deadline across workspace and phased graph admission', async () => { + it('applies no-wait and timeouts to requests admitted after a shared iteration starts', async () => { jest.useFakeTimers(); const graphStarted = createDeferred(); const releaseGraph = createDeferred(); @@ -379,13 +380,8 @@ describe('request admission integration', () => { createPhasedRequest('first-phased'), new TestPhasedRequestClient() ); - const timeoutClient: TestPhasedRequestClient = new TestPhasedRequestClient(); - const timeoutPhased = phasedRouter.executeAsync( - createPhasedRequest('timeout-phased', { waitTimeoutMs: 10 }), - timeoutClient - ); - jest.advanceTimersByTime(8); releaseWorkspace.resolve(); + await jest.advanceTimersByTimeAsync(0); await graphStarted.promise; const noWaitResult = await phasedRouter.executeAsync( @@ -393,18 +389,19 @@ describe('request admission integration', () => { new TestPhasedRequestClient() ); expect(noWaitResult).toMatchObject({ admissionErrorCode: 'no-wait', outcome: 'failure' }); + const timeoutClient: TestPhasedRequestClient = new TestPhasedRequestClient(); + const timeoutPhased = phasedRouter.executeAsync( + createPhasedRequest('timeout-phased', { waitTimeoutMs: 10 }), + timeoutClient + ); + await jest.advanceTimersByTimeAsync(10); + const timeoutResult = await timeoutPhased; + expect(timeoutResult).toMatchObject({ admissionErrorCode: 'wait-timeout', outcome: 'failure' }); expect( timeoutClient.writes .map(({ queuePosition }) => queuePosition?.payload.position) .filter((position): position is number => position !== undefined) - ).toEqual(expect.arrayContaining([2, 1])); - jest.advanceTimersByTime(2); - const timeoutResult = await timeoutPhased; - expect(timeoutResult).toMatchObject({ - admissionErrorCode: 'wait-timeout', - errorMessage: 'The request was not admitted within 10ms.', - outcome: 'failure' - }); + ).toContain(1); expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(1); releaseGraph.resolve(); @@ -443,6 +440,69 @@ describe('request admission integration', () => { expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(1); }); + it('keeps an exclusive FIFO gate between shared-build batches', async () => { + const buildStarted = createDeferred(); + const releaseBuild = createDeferred(); + const fixture = createRoutingFixture( + new Map([ + [ + TEST_OPERATION, + new TestOperationRunner(TEST_OPERATION, OperationStatus.Success, async (): Promise => { + buildStarted.resolve(); + await releaseBuild.promise; + }) + ] + ]) + ); + const globalRouter: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(fixture.session); + const phasedRouter: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const firstBuild = phasedRouter.executeAsync( + { + commandName: 'build', + commandOrigin: 'built-in', + engineShape: TEST_ENGINE_SHAPE, + environment: {}, + operationSelection: [{ enabledState: true, operationId: TEST_OPERATION }], + requestId: 'first-build' + }, + new TestPhasedRequestClient('first-build') + ); + await buildStarted.promise; + + const exclusiveStarted = createDeferred(); + const releaseExclusive = createDeferred(); + const exclusive = globalRouter.executeAsync( + createRequest(globalRouter, 'exclusive', 'custom-exclusive'), + createBlockingExecutor(exclusiveStarted.resolve, releaseExclusive.promise), + new AdmissionClient() + ); + let lateBuildSettled: boolean = false; + const lateBuild = phasedRouter + .executeAsync( + { + commandName: 'build', + commandOrigin: 'built-in', + engineShape: TEST_ENGINE_SHAPE, + environment: {}, + operationSelection: [{ enabledState: true, operationId: TEST_OPERATION }], + requestId: 'late-build' + }, + new TestPhasedRequestClient('late-build') + ) + .then((result) => { + lateBuildSettled = true; + return result; + }); + + releaseBuild.resolve(); + await exclusiveStarted.promise; + expect(lateBuildSettled).toBe(false); + releaseExclusive.resolve(); + + await Promise.all([firstBuild, exclusive, lateBuild]); + expect(lateBuildSettled).toBe(true); + }); + it('holds an exclusive lease until cleanup and final-result output settle', async () => { const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( new TestWorkspaceSession(TEST_REPO_ROOT) From d8e25f49f9f38aa598df7e4c6e7a64d0274cd8ec Mon Sep 17 00:00:00 2001 From: mojaza Date: Wed, 26 Aug 2026 14:27:41 -0700 Subject: [PATCH 2/4] Fix shared-build test client replay Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/test/PhasedRequestRouterTestUtilities.ts | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts index 8b500de4a3..c61dc8e33d 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts @@ -71,12 +71,14 @@ export class TestPhasedRequestClient implements IPhasedRequestClient { public onWriteAsync: ((write: ITestClientWrite) => Promise) | undefined; readonly #sequenceState: { next: number }; - public constructor(sequenceState: { next: number } = { next: 1 }) { - this.#sequenceState = sequenceState; - } - - public constructor(sessionId: string = 'test-session') { - this.sessionId = sessionId; + public constructor(sessionIdOrSequenceState: string | { next: number } = 'test-session') { + if (typeof sessionIdOrSequenceState === 'string') { + this.sessionId = sessionIdOrSequenceState; + this.#sequenceState = { next: 1 }; + } else { + this.sessionId = 'test-session'; + this.#sequenceState = sessionIdOrSequenceState; + } } public get abortSignal(): AbortSignal { From a598894a209a1a700bd35b744033f07fd0143cc3 Mon Sep 17 00:00:00 2001 From: mojaza Date: Wed, 26 Aug 2026 14:49:47 -0700 Subject: [PATCH 3/4] Fix shared-build admission and selection precedence Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../rush-daemon/src/PhasedRequestRouter.ts | 62 +++++++++++++------ .../src/test/PhasedRequestBatching.test.ts | 57 +++++++++++++++++ .../test/RequestAdmissionIntegration.test.ts | 3 +- 3 files changed, 103 insertions(+), 19 deletions(-) diff --git a/libraries/rush-daemon/src/PhasedRequestRouter.ts b/libraries/rush-daemon/src/PhasedRequestRouter.ts index 990ca2a1d8..ea2426bc01 100644 --- a/libraries/rush-daemon/src/PhasedRequestRouter.ts +++ b/libraries/rush-daemon/src/PhasedRequestRouter.ts @@ -210,6 +210,7 @@ class PhasedRequestBatchCoordinator { #acceptingCurrentBatch: boolean = false; #currentBatch: ReadonlyArray | undefined; #drainScheduled: boolean = false; + #nextGraphLeasePromise: Promise | undefined; #running: boolean = false; public constructor( @@ -265,6 +266,9 @@ class PhasedRequestBatchCoordinator { return; } this.#drainScheduled = true; + this.#nextGraphLeasePromise = this.#graphExecutionScheduler.acquireAsync({ + exclusivityClass: RequestExclusivityClass.Exclusive + }); setImmediate(() => { this.#drainScheduled = false; void this.#drainAsync(); @@ -277,6 +281,13 @@ class PhasedRequestBatchCoordinator { } this.#running = true; try { + if (this.#pending.length === 0) { + const unusedGraphLeasePromise: Promise | undefined = + this.#nextGraphLeasePromise; + this.#nextGraphLeasePromise = undefined; + (await unusedGraphLeasePromise)?.release(); + return; + } while (this.#pending.length > 0) { const first: IBatchEntry = this.#pending.shift()!; const batch: IBatchEntry[] = [first]; @@ -307,12 +318,14 @@ class PhasedRequestBatchCoordinator { } #canJoinCurrentBatch(request: IPreparedPhasedRequest): boolean { + if (request.exclusivityClass !== RequestExclusivityClass.SharedBuild) { + return false; + } if (!this.#running) { return true; } return ( this.#acceptingCurrentBatch && - request.exclusivityClass === RequestExclusivityClass.SharedBuild && this.#currentBatch?.[0]?.exclusivityClass === RequestExclusivityClass.SharedBuild ); } @@ -331,9 +344,12 @@ class PhasedRequestBatchCoordinator { } async #executeBatchAsync(batch: IBatchEntry[]): Promise { - const graphLeasePromise: Promise = this.#graphExecutionScheduler.acquireAsync({ - exclusivityClass: RequestExclusivityClass.Exclusive - }); + const graphLeasePromise: Promise = + this.#nextGraphLeasePromise ?? + this.#graphExecutionScheduler.acquireAsync({ + exclusivityClass: RequestExclusivityClass.Exclusive + }); + this.#nextGraphLeasePromise = undefined; const graphLease: IRequestLease = await graphLeasePromise; try { if (this.#graph.hasScheduledIteration || this.#graph.status === OperationStatus.Executing) { @@ -743,23 +759,33 @@ function applySelections( graph: IOperationGraph, selections: ReadonlyArray ): void { - graph.setEnabledStates(graph.operations, false, 'unsafe'); - graph.setEnabledStates( - selections.flatMap( - (selection: IResolvedSelection) => selection.ignoreDependencyOperations - ), - 'ignore-dependency-changes', - 'safe' + const enabledOperations: ReadonlyArray = selections.flatMap( + (selection: IResolvedSelection) => selection.enabledOperations ); - graph.setEnabledStates( - selections.flatMap((selection: IResolvedSelection) => selection.enabledOperations), - true, - 'safe' + const ignoreDependencyOperations: ReadonlyArray = selections.flatMap( + (selection: IResolvedSelection) => selection.ignoreDependencyOperations + ); + const enabledClosureBySelection: ReadonlyArray> = selections.map( + (selection: IResolvedSelection) => + new Set(collectSelectionClosure(selection.enabledOperations, [])) ); + const effectiveIgnoreDependencyOperations: Operation[] = []; + selections.forEach((selection: IResolvedSelection, selectionIndex: number) => { + for (const operation of selection.ignoreDependencyOperations) { + const requiredByAnotherSelection: boolean = enabledClosureBySelection.some( + (enabledClosure: ReadonlySet, enabledSelectionIndex: number) => + enabledSelectionIndex !== selectionIndex && enabledClosure.has(operation) + ); + if (!requiredByAnotherSelection) { + effectiveIgnoreDependencyOperations.push(operation); + } + } + }); + graph.setEnabledStates(graph.operations, false, 'unsafe'); + graph.setEnabledStates(ignoreDependencyOperations, 'ignore-dependency-changes', 'safe'); + graph.setEnabledStates(enabledOperations, true, 'safe'); graph.setEnabledStates( - selections.flatMap( - (selection: IResolvedSelection) => selection.ignoreDependencyOperations - ), + effectiveIgnoreDependencyOperations, 'ignore-dependency-changes', 'unsafe' ); diff --git a/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts b/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts index a21d17ca13..a7f2b90a8f 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts @@ -364,6 +364,63 @@ describe('shared phased request batching', () => { expect(scheduleSpy).toHaveBeenCalledTimes(3); }); + it('applies graph admission to same-turn shared-read requests', async () => { + const operationStarted: IDeferred = createDeferred(); + const releaseOperation: IDeferred = createDeferred(); + const fixture: ITestRoutingFixture = createFixture({ + actionAAsync: async (): Promise => { + operationStarted.resolve(); + await releaseOperation.promise; + } + }); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const first = router.executeAsync( + { ...createRequest('first', OPERATION_A), commandName: 'list' }, + new TestPhasedRequestClient('one') + ); + const noWait = router.executeAsync( + { + ...createRequest('no-wait', OPERATION_C), + admission: { noWait: true }, + commandName: 'list' + }, + new TestPhasedRequestClient('two') + ); + + const noWaitResult: IDaemonPhasedRequestResult = await noWait; + expect(noWaitResult).toMatchObject({ admissionErrorCode: 'no-wait', outcome: 'failure' }); + await operationStarted.promise; + releaseOperation.resolve(); + await first; + }); + + it('keeps true enabled state dominant across merged selections', async () => { + const fixture: ITestRoutingFixture = createFixture(); + const enabledStates: Array = []; + fixture.graph.hooks.onIterationScheduled.tap('capture enabled state', () => { + enabledStates.push(fixture.operations.get(OPERATION_A)?.enabled); + }); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + + await Promise.all([ + router.executeAsync( + { + ...createRequest('ignore-dependency', OPERATION_A), + operationSelection: [ + { enabledState: 'ignore-dependency-changes', operationId: OPERATION_A } + ] + }, + new TestPhasedRequestClient('one') + ), + router.executeAsync( + createRequest('requires-dependency', OPERATION_B), + new TestPhasedRequestClient('two') + ) + ]); + + expect(enabledStates).toEqual([true]); + }); + it('preserves per-client backpressure and final-result ordering in a merged batch', async () => { const fixture: ITestRoutingFixture = createFixture({ actionAAsync: async (terminal: ITerminal): Promise => { diff --git a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts index 382f0139e5..fb0f0e31fe 100644 --- a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts +++ b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts @@ -116,13 +116,14 @@ function createPhasedRequest( } function createLegacyPhasedRequest(requestId: string): IDaemonPhasedRequest { - return { + const legacyRequest: Partial = { commandName: 'build', engineShape: TEST_ENGINE_SHAPE, environment: {}, operationSelection: [{ enabledState: true, operationId: TEST_OPERATION }], requestId }; + return legacyRequest as IDaemonPhasedRequest; } function createDeferred(): { readonly promise: Promise; readonly resolve: () => void } { From d5c531cf22c09d6545493ced11ff468ef1465b1d Mon Sep 17 00:00:00 2001 From: mojaza Date: Thu, 27 Aug 2026 13:40:05 -0700 Subject: [PATCH 4/4] [rush-daemon] Address WS2.9 review feedback Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/PhasedRequestEventMultiplexer.ts | 15 ++++- .../rush-daemon/src/PhasedRequestEventSink.ts | 23 +++++++- .../rush-daemon/src/PhasedRequestRouter.ts | 28 ++++++--- .../src/test/PhasedRequestBatching.test.ts | 59 +++++++++++++++++++ 4 files changed, 113 insertions(+), 12 deletions(-) diff --git a/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts b/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts index a6271bfaed..3ac90ca137 100644 --- a/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts +++ b/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts @@ -9,15 +9,19 @@ import type { } from '@microsoft/rush-lib'; import type { ITerminalChunk } from '@rushstack/terminal'; +interface IRequestEventSink extends _IOperationGraphEventSink { + onIterationScheduled(records: Iterable): void; +} + export class PhasedRequestEventMultiplexer implements _IOperationGraphEventSink { readonly #workspaceSink: _IOperationGraphEventSink | undefined; - readonly #requestSinks: Set<_IOperationGraphEventSink> = new Set(); + readonly #requestSinks: Set = new Set(); public constructor(workspaceSink: _IOperationGraphEventSink | undefined) { this.#workspaceSink = workspaceSink; } - public subscribe(requestSink: _IOperationGraphEventSink): () => void { + public subscribe(requestSink: IRequestEventSink): () => void { this.#requestSinks.add(requestSink); let subscribed: boolean = true; return () => { @@ -28,6 +32,13 @@ export class PhasedRequestEventMultiplexer implements _IOperationGraphEventSink }; } + public onIterationScheduled(records: Iterable): void { + const executionResults: IOperationExecutionResult[] = [...records]; + for (const requestSink of this.#requestSinks) { + requestSink.onIterationScheduled(executionResults); + } + } + public onOperationRegistered(operationId: string, silent: boolean): void { this.#workspaceSink?.onOperationRegistered?.(operationId, silent); for (const requestSink of this.#requestSinks) { diff --git a/libraries/rush-daemon/src/PhasedRequestEventSink.ts b/libraries/rush-daemon/src/PhasedRequestEventSink.ts index 1650bd1777..d25b936da4 100644 --- a/libraries/rush-daemon/src/PhasedRequestEventSink.ts +++ b/libraries/rush-daemon/src/PhasedRequestEventSink.ts @@ -31,7 +31,7 @@ const TEXT_ENCODER: InstanceType = new TextEncoder(); interface IObservedOperationResult { readonly executionResult: IOperationExecutionResult; - readonly status: string; + readonly status: OperationStatus; } interface IEventOptions { @@ -91,6 +91,8 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink { readonly #observedResults: Map = new Map(); readonly #rushVersion: string; readonly #writer: OrderedClientWriter; + #completedOperations: number = 0; + #totalOperations: number = 0; public constructor(options: { activeOperationIds: ReadonlySet; @@ -120,6 +122,16 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink { } } + public onIterationScheduled(records: Iterable): void { + this.#completedOperations = 0; + this.#totalOperations = 0; + for (const record of records) { + if (this.#activeOperationIds.has(record.operation.name) && !record.silent) { + this.#totalOperations++; + } + } + } + public onOperationStatusChanged( result: IOperationExecutionResult, previousStatus: OperationStatus @@ -139,12 +151,17 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink { }); } - public onOperationHeader(operationId: string, completed: number, total: number): void { + public onOperationHeader(operationId: string): void { if (this.#activeOperationIds.has(operationId)) { + this.#completedOperations++; this.#emitEvent( 'extension', { - data: { completedOperations: completed, operationId, totalOperations: total }, + data: { + completedOperations: this.#completedOperations, + operationId, + totalOperations: this.#totalOperations + }, name: RUSHD_OPERATION_HEADER }, { required: true } diff --git a/libraries/rush-daemon/src/PhasedRequestRouter.ts b/libraries/rush-daemon/src/PhasedRequestRouter.ts index ea2426bc01..bc21068abd 100644 --- a/libraries/rush-daemon/src/PhasedRequestRouter.ts +++ b/libraries/rush-daemon/src/PhasedRequestRouter.ts @@ -83,6 +83,11 @@ interface IBatchEntry extends IPreparedPhasedRequest { } const ROUTING_STATE_BY_GRAPH: WeakMap = new WeakMap(); +const OBSERVED_STATUS_OVERRIDES_RETAINED: ReadonlySet = new Set([ + OperationStatus.Aborted, + OperationStatus.Blocked, + OperationStatus.Skipped +]); /** * Routes one caller-resolved phased request through a real warm workspace operation graph. @@ -607,6 +612,9 @@ function getGraphRoutingState( const multiplexer: PhasedRequestEventMultiplexer = new PhasedRequestEventMultiplexer( getGraphEventSink(graph) ); + graph.hooks.onIterationScheduled.tap('rushd request event multiplexer', (records) => { + multiplexer.onIterationScheduled(records.values()); + }); const graphExecutionScheduler: RequestScheduler = new RequestScheduler(); state = { coordinator: new PhasedRequestBatchCoordinator( @@ -802,16 +810,22 @@ function collectOperationOutcomes( const observed: ReturnType = requestSink.getObservedResult(operation); const retained: IOperationExecutionResult | undefined = graph.resultByOperation.get(operation); - const status: string | undefined = - retained?.status ?? - observed?.status ?? - (fillMissingAsAborted ? OperationStatus.Aborted : undefined); + let status: string | undefined; + let errorMessage: string | undefined; + if ( + observed !== undefined && + (retained === undefined || OBSERVED_STATUS_OVERRIDES_RETAINED.has(observed.status)) + ) { + status = observed.status; + errorMessage = observed.executionResult.error?.message; + } else { + status = retained?.status ?? observed?.status; + errorMessage = retained?.error?.message ?? observed?.executionResult.error?.message; + } + status ??= fillMissingAsAborted ? OperationStatus.Aborted : undefined; if (status === undefined) { continue; } - const errorMessage: string | undefined = observed - ? observed.executionResult.error?.message - : retained?.error?.message; outcomes.push({ observedInCurrentIteration: observed !== undefined, result: { operationId: operation.name, status, errorMessage }, diff --git a/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts b/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts index a7f2b90a8f..9ed2adef1c 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestBatching.test.ts @@ -8,6 +8,7 @@ import type { IDaemonPhasedRequest, IDaemonPhasedRequestResult } from '@rushstack/rush-daemon-protocol'; +import { RUSHD_OPERATION_HEADER } from '@rushstack/rush-daemon-protocol'; import { OperationStatus } from '@microsoft/rush-lib'; import { PhasedRequestRouter } from '../PhasedRequestRouter'; @@ -136,6 +137,12 @@ describe('shared phased request batching', () => { expect(getResultOperationIds(resultC)).toEqual([OPERATION_C]); expect(getWrittenOperationIds(clientA)).toEqual(new Set([OPERATION_A])); expect(getWrittenOperationIds(clientC)).toEqual(new Set([OPERATION_C])); + expect(getHeaderData(clientA)).toEqual([ + { completedOperations: 1, operationId: OPERATION_A, totalOperations: 1 } + ]); + expect(getHeaderData(clientC)).toEqual([ + { completedOperations: 1, operationId: OPERATION_C, totalOperations: 1 } + ]); }); it('derives shared and disjoint failure results from each client subset', async () => { @@ -240,6 +247,33 @@ describe('shared phased request batching', () => { expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(1); }); + it('prefers a current abort after invalidating a retained warm success', async () => { + const fixture: ITestRoutingFixture = createFixture(); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const first = await router.executeAsync( + createRequest('first', OPERATION_A), + new TestPhasedRequestClient('one') + ); + expect(first.operationResults).toEqual([ + expect.objectContaining({ operationId: OPERATION_A, status: OperationStatus.Success }) + ]); + fixture.graph.invalidateOperations(undefined, 'rerun'); + fixture.graph.hooks.beforeExecuteIterationAsync.tapPromise( + 'abort current iteration', + async (): Promise => OperationStatus.Aborted + ); + + const second = await router.executeAsync( + createRequest('second', OPERATION_A), + new TestPhasedRequestClient('two') + ); + + expect(second).toMatchObject({ exitCode: 1, outcome: 'aborted', scheduled: true }); + expect(second.operationResults).toEqual([ + expect.objectContaining({ operationId: OPERATION_A, status: OperationStatus.Aborted }) + ]); + }); + it('preserves failure precedence when a client cancels during a failing shared operation', async () => { const operationStarted: IDeferred = createDeferred(); const releaseOperation: IDeferred = createDeferred(); @@ -498,3 +532,28 @@ function getWrittenOperationIds(client: TestPhasedRequestClient): ReadonlySet { + return client.writes.flatMap(({ event }) => { + const payload: unknown = event?.payload; + if ( + typeof payload !== 'object' || + payload === null || + (payload as { name?: unknown }).name !== RUSHD_OPERATION_HEADER + ) { + return []; + } + const data: unknown = (payload as { data?: unknown }).data; + return typeof data === 'object' && data !== null + ? [ + data as { + completedOperations: number; + operationId: string; + totalOperations: number; + } + ] + : []; + }); +}