diff --git a/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-phased-request-contracts_2026-08-21-17-24.json b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-phased-request-contracts_2026-08-21-17-24.json new file mode 100644 index 0000000000..cd90412e6c --- /dev/null +++ b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-phased-request-contracts_2026-08-21-17-24.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon-protocol", + "comment": "Add typed resolved phased-request, enabled-state selection, engine-shape, and client-scoped operation result contracts.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon-protocol", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-daemon/mojazayeri-route-phased-requests_2026-08-21-17-24.json b/common/changes/@rushstack/rush-daemon/mojazayeri-route-phased-requests_2026-08-21-17-24.json new file mode 100644 index 0000000000..6eface8a32 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-route-phased-requests_2026-08-21-17-24.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon", + "comment": "Add an opt-in phased request router that validates a caller-resolved selection, reconciles warm invalidations, runs one real graph iteration, scopes ordered streams and events to the client, and safely aborts on cancellation or disconnect.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-terminal-renderer/mojazayeri-authoritative-operation-headers_2026-08-25-13-53.json b/common/changes/@rushstack/rush-terminal-renderer/mojazayeri-authoritative-operation-headers_2026-08-25-13-53.json new file mode 100644 index 0000000000..c80e4784d5 --- /dev/null +++ b/common/changes/@rushstack/rush-terminal-renderer/mojazayeri-authoritative-operation-headers_2026-08-25-13-53.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-terminal-renderer", + "comment": "Use authoritative daemon operation-header counters when collating partial warm iterations.", + "type": "patch" + } + ], + "packageName": "@rushstack/rush-terminal-renderer", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/config/subspaces/default/pnpm-lock.yaml b/common/config/subspaces/default/pnpm-lock.yaml index f6f0ed902f..1194185085 100644 --- a/common/config/subspaces/default/pnpm-lock.yaml +++ b/common/config/subspaces/default/pnpm-lock.yaml @@ -4103,6 +4103,9 @@ importers: '@rushstack/rush-daemon-transport': specifier: workspace:* version: link:../rush-daemon-transport + '@rushstack/terminal': + specifier: workspace:* + version: link:../terminal devDependencies: '@rushstack/heft': specifier: workspace:* diff --git a/common/reviews/api/rush-daemon-protocol.api.md b/common/reviews/api/rush-daemon-protocol.api.md index 37834b8ba5..9f406ac255 100644 --- a/common/reviews/api/rush-daemon-protocol.api.md +++ b/common/reviews/api/rush-daemon-protocol.api.md @@ -100,6 +100,9 @@ export type DaemonJsonValue = string | number | boolean | DaemonJsonNull | reado readonly [key: string]: DaemonJsonValue; }; +// @beta +export type DaemonPhasedOperationEnabledState = true | 'ignore-dependency-changes'; + // @beta export class DaemonProtocolError extends Error { constructor(code: DaemonProtocolErrorCode, message: string, options?: IDaemonProtocolErrorOptions); @@ -278,6 +281,41 @@ export interface IDaemonOperationStreamClosedPayload { readonly operationId: string; } +// @beta +export interface IDaemonPhasedEngineShape { + readonly phaseNames: ReadonlyArray; + readonly pluginNames: ReadonlyArray; +} + +// @beta +export interface IDaemonPhasedOperationResult { + readonly errorMessage?: string; + readonly operationId: string; + readonly status: string; +} + +// @beta +export interface IDaemonPhasedOperationSelection { + readonly enabledState: DaemonPhasedOperationEnabledState; + readonly operationId: string; +} + +// @beta +export interface IDaemonPhasedRequest { + readonly commandName: string; + readonly engineShape: IDaemonPhasedEngineShape; + readonly operationSelection: ReadonlyArray; + readonly requestId: string; +} + +// @beta +export interface IDaemonPhasedRequestResult { + readonly aborted: boolean; + readonly operationResults: ReadonlyArray; + readonly requestId: string; + readonly scheduled: boolean; +} + // @beta export interface IDaemonPingMessage { // (undocumented) diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index 9277421b0e..cec673082f 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -7,7 +7,10 @@ /// import type { GetInputsSnapshotAsyncFn } from '@microsoft/rush-lib'; +import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol'; import type { IDaemonPaths } from '@rushstack/rush-daemon-transport'; +import type { IDaemonPhasedRequest } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonPhasedRequestResult } from '@rushstack/rush-daemon-protocol'; import type { IInputsSnapshot } from '@microsoft/rush-lib'; import type { IOperationGraph } from '@microsoft/rush-lib'; import type { Operation } from '@microsoft/rush-lib'; @@ -58,6 +61,15 @@ export interface IMapWorkspaceInvalidationsOptions { readonly operationGraph: IOperationGraph; } +// @beta +export interface IPhasedRequestClient { + readonly abortSignal: AbortSignal; + getNextEventSequence(): number; + readonly sessionId: string; + writeEventAsync(event: IDaemonEventEnvelope): Promise; + writeLogChunkAsync(operationId: string, stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; +} + // @public export interface IRequestLease { // (undocumented) @@ -217,6 +229,12 @@ export interface IWorkspaceSessionOptions { // @beta export type MapWorkspaceInvalidationsToOperationsAsync = (options: IMapWorkspaceInvalidationsOptions) => Promise>; +// @beta +export class PhasedRequestRouter { + constructor(workspaceSession: IWorkspaceSession); + executeAsync(request: IDaemonPhasedRequest, client: IPhasedRequestClient): Promise; +} + // @public export enum RequestExclusivityClass { // (undocumented) diff --git a/common/reviews/api/rush-terminal-renderer.api.md b/common/reviews/api/rush-terminal-renderer.api.md index e13568ec40..c136aa1aed 100644 --- a/common/reviews/api/rush-terminal-renderer.api.md +++ b/common/reviews/api/rush-terminal-renderer.api.md @@ -7,6 +7,7 @@ import type { DaemonVerbosity } from '@rushstack/rush-daemon-protocol'; import type { IDaemonClientCaps } from '@rushstack/rush-daemon-protocol'; import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonOperationHeaderPayload } from '@rushstack/rush-daemon-protocol'; import { ITerminalChunk } from '@rushstack/terminal'; import { TerminalWritable } from '@rushstack/terminal'; @@ -82,6 +83,7 @@ export class OperationStreamRegistry { constructor(options: IOperationStreamRegistryOptions); closeOperation(operationId: string): void; registerOperation(): void; + setOperationHeader(header: IDaemonOperationHeaderPayload): void; writeChunk(operationId: string, chunk: ITerminalChunk): void; } diff --git a/libraries/rush-daemon-protocol/README.md b/libraries/rush-daemon-protocol/README.md index 935b7f65a2..96a59894e1 100644 --- a/libraries/rush-daemon-protocol/README.md +++ b/libraries/rush-daemon-protocol/README.md @@ -17,6 +17,9 @@ The engine-agnostic **wire layer** spoken by every client of the Rush daemon (`r reference when the reporter package lands) plus namespaced `rushd.*` extension events. - **Per-subscription verbosity** — a pure filter applied at event serialization so each client receives its own verbosity subset without mutating shared engine state. +- **Resolved phased-request contracts** — engine-agnostic request, enabled-state selection, + and client-scoped result types for integrations that have already parsed a command and + resolved it against a real warm operation graph. Part of the Rush 6 / rushd re-architecture: [microsoft/rushstack#5894](https://github.com/microsoft/rushstack/issues/5894). diff --git a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts new file mode 100644 index 0000000000..dc0ee92028 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts @@ -0,0 +1,83 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** + * The enabled state assigned to one selected operation by a phased request. + * + * @beta + */ +export type DaemonPhasedOperationEnabledState = true | 'ignore-dependency-changes'; + +/** + * One caller-resolved operation selection. + * + * @remarks + * Operation identifiers come from the integration-owned real operation graph. Command-line parsing and graph + * construction remain outside the wire contract. + * + * @beta + */ +export interface IDaemonPhasedOperationSelection { + /** The non-disabled state to apply with the real graph's enabled-state API. */ + readonly enabledState: DaemonPhasedOperationEnabledState; + /** The integration-resolved operation identifier. */ + readonly operationId: string; +} + +/** + * The explicit phase and plugin shape of the warm graph used by a phased request. + * + * @beta + */ +export interface IDaemonPhasedEngineShape { + /** Every phase represented by the warm graph. */ + readonly phaseNames: ReadonlyArray; + /** Every plugin applied when the warm graph was constructed. */ + readonly pluginNames: ReadonlyArray; +} + +/** + * A typed phased request after an integration has parsed the command and resolved its operation selection. + * + * @beta + */ +export interface IDaemonPhasedRequest { + /** The parsed phased command name. */ + readonly commandName: string; + /** The exact warm engine shape against which the selection was resolved. */ + readonly engineShape: IDaemonPhasedEngineShape; + /** The caller-resolved selected operations and their enabled states. */ + readonly operationSelection: ReadonlyArray; + /** A client-generated identifier unique within the connection. */ + readonly requestId: string; +} + +/** + * The client-scoped result for one selected operation. + * + * @beta + */ +export interface IDaemonPhasedOperationResult { + /** The operation identifier used by the request and its streamed output. */ + readonly operationId: string; + /** The raw Rush operation status. */ + readonly status: string; + /** The operation error message, when execution produced one. */ + readonly errorMessage?: string; +} + +/** + * The result of routing one phased request through a warm operation graph. + * + * @beta + */ +export interface IDaemonPhasedRequestResult { + /** Whether cancellation or disconnect aborted the iteration. */ + readonly aborted: boolean; + /** Results only for operations enabled for this client. */ + readonly operationResults: ReadonlyArray; + /** The identifier copied from the request. */ + readonly requestId: string; + /** Whether the real graph scheduled work for this iteration. */ + readonly scheduled: boolean; +} diff --git a/libraries/rush-daemon-protocol/src/index.ts b/libraries/rush-daemon-protocol/src/index.ts index 4ef7a7107c..878386d550 100644 --- a/libraries/rush-daemon-protocol/src/index.ts +++ b/libraries/rush-daemon-protocol/src/index.ts @@ -50,3 +50,11 @@ export { decodeDaemonEventFrame, encodeDaemonEventFrame, serializeDaemonEventFor export type { IDaemonActivityPayload, IDaemonOperationRegisteredPayload, IDaemonOperationStatusChangedPayload } from './DaemonOperationPayloads'; export { RUSHD_OPERATION_HEADER, RUSHD_OPERATION_STREAM_CLOSED } from './DaemonRushdExtensions'; export type { IDaemonExtensionEventPayload, IDaemonOperationHeaderPayload, IDaemonOperationStreamClosedPayload } from './DaemonRushdExtensions'; +export type { + DaemonPhasedOperationEnabledState, + IDaemonPhasedEngineShape, + IDaemonPhasedOperationResult, + IDaemonPhasedOperationSelection, + IDaemonPhasedRequest, + IDaemonPhasedRequestResult +} from './DaemonPhasedRequest'; diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index f8e6cab2db..adc605597d 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -27,3 +27,16 @@ no paths to classify and therefore remains a full invalidation. The routing laye workspace session rather than run a stale graph. The default daemon executable does not construct or route this graph while the command-independent plugin shape and per-iteration runner lifetime tracked by [rushstack#5895](https://github.com/microsoft/rushstack/issues/5895) remain incomplete. + +`PhasedRequestRouter` is the opt-in execution boundary once an integration has supplied that real warm graph. The +integration parses the command and supplies an explicit phase/plugin shape plus operation enabled-state selection; +the router validates both, reconciles retained invalidations, applies the selection with `IOperationGraph.setEnabledStates`, +and runs at most one scheduled iteration. Requests are serialized until shared-build merging is implemented. A +requesting client receives only its enabled dependency closure's WS1 raw chunks and structured events through +backpressured, ordered callbacks, followed by client-scoped operation results. Cancellation or disconnect aborts the +current iteration without closing daemon-owned runners or the graph. + +This layer deliberately does not add control-frame admission or 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; full command parsing remains blocked by +[rushstack#5895](https://github.com/microsoft/rushstack/issues/5895). diff --git a/libraries/rush-daemon/package.json b/libraries/rush-daemon/package.json index eb628e4a5e..89ff16cf80 100644 --- a/libraries/rush-daemon/package.json +++ b/libraries/rush-daemon/package.json @@ -48,7 +48,8 @@ "@microsoft/rush-lib": "workspace:*", "@rushstack/node-core-library": "workspace:*", "@rushstack/rush-daemon-protocol": "workspace:*", - "@rushstack/rush-daemon-transport": "workspace:*" + "@rushstack/rush-daemon-transport": "workspace:*", + "@rushstack/terminal": "workspace:*" }, "devDependencies": { "@rushstack/heft": "workspace:*", diff --git a/libraries/rush-daemon/src/PhasedRequestClient.ts b/libraries/rush-daemon/src/PhasedRequestClient.ts new file mode 100644 index 0000000000..68dfc72a6f --- /dev/null +++ b/libraries/rush-daemon/src/PhasedRequestClient.ts @@ -0,0 +1,33 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol'; + +/** + * A client-scoped destination for one routed phased request. + * + * @remarks + * The client must abort `abortSignal` when its request is cancelled or its connection closes. Writes are invoked + * serially in engine order; each promise provides the destination's backpressure boundary. + * + * @beta + */ +export interface IPhasedRequestClient { + /** Aborted by the transport when the request is cancelled or disconnected. */ + readonly abortSignal: AbortSignal; + /** The connection session identifier used in structured event envelopes. */ + readonly sessionId: string; + + /** Returns the next structured-event sequence number for this connection. */ + getNextEventSequence(): number; + + /** Writes one structured event through the client's backpressured destination. */ + writeEventAsync(event: IDaemonEventEnvelope): Promise; + + /** Writes one operation-scoped output chunk through the client's backpressured destination. */ + writeLogChunkAsync( + operationId: string, + stream: 'stdout' | 'stderr', + chunk: Uint8Array + ): Promise; +} diff --git a/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts b/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts new file mode 100644 index 0000000000..431d476cc8 --- /dev/null +++ b/libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts @@ -0,0 +1,68 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IOperationExecutionResult, + OperationStatus, + _IOperationActivityOptions, + _IOperationGraphEventSink +} from '@microsoft/rush-lib'; +import type { ITerminalChunk } from '@rushstack/terminal'; + +export class PhasedRequestEventMultiplexer implements _IOperationGraphEventSink { + readonly #workspaceSink: _IOperationGraphEventSink | undefined; + #requestSink: _IOperationGraphEventSink | undefined; + + 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; + let subscribed: boolean = true; + return () => { + if (subscribed) { + subscribed = false; + if (this.#requestSink === requestSink) { + this.#requestSink = undefined; + } + } + }; + } + + public onOperationRegistered(operationId: string, silent: boolean): void { + this.#workspaceSink?.onOperationRegistered?.(operationId, silent); + this.#requestSink?.onOperationRegistered?.(operationId, silent); + } + + public onOperationStatusChanged( + result: IOperationExecutionResult, + previousStatus: OperationStatus + ): void { + this.#workspaceSink?.onOperationStatusChanged?.(result, previousStatus); + this.#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); + } + + public onOperationChunk(operationId: string, chunk: ITerminalChunk): void { + this.#workspaceSink?.onOperationChunk?.(operationId, chunk); + this.#requestSink?.onOperationChunk?.(operationId, chunk); + } + + public onOperationStreamClosed(operationId: string): void { + this.#workspaceSink?.onOperationStreamClosed?.(operationId); + this.#requestSink?.onOperationStreamClosed?.(operationId); + } + + public onActivity(text: string, options?: _IOperationActivityOptions): void { + this.#workspaceSink?.onActivity?.(text, options); + this.#requestSink?.onActivity?.(text, options); + } +} diff --git a/libraries/rush-daemon/src/PhasedRequestEventSink.ts b/libraries/rush-daemon/src/PhasedRequestEventSink.ts new file mode 100644 index 0000000000..a2f137bbb0 --- /dev/null +++ b/libraries/rush-daemon/src/PhasedRequestEventSink.ts @@ -0,0 +1,208 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { randomUUID } from 'node:crypto'; + +import type { + IOperationExecutionResult, + Operation, + OperationStatus, + _IOperationActivityOptions, + _IOperationGraphEventSink +} from '@microsoft/rush-lib'; +import { + DAEMON_PROTOCOL_VERSION, + RUSHD_OPERATION_HEADER, + RUSHD_OPERATION_STREAM_CLOSED +} from '@rushstack/rush-daemon-protocol'; +import type { + DaemonEventType, + IDaemonEventEnvelope, + IDaemonEventScope +} from '@rushstack/rush-daemon-protocol'; +import { TerminalChunkKind } from '@rushstack/terminal'; +import type { ITerminalChunk } from '@rushstack/terminal'; + +import type { IPhasedRequestClient } from './PhasedRequestClient'; + +const EVENT_SOURCE_PACKAGE: string = '@microsoft/rush-lib'; +const EVENT_SOURCE_COMPONENT: string = 'OperationGraph'; +const TEXT_ENCODER: InstanceType = new TextEncoder(); + +interface IObservedOperationResult { + readonly executionResult: IOperationExecutionResult; + readonly status: string; +} + +interface IEventOptions { + readonly required?: boolean; + readonly scope?: IDaemonEventScope; +} + +class OrderedClientWriter { + readonly #client: IPhasedRequestClient; + readonly #onFailure: () => void; + #failure: Error | undefined; + #tail: Promise = Promise.resolve(); + + public constructor(client: IPhasedRequestClient, onFailure: () => void) { + this.#client = client; + this.#onFailure = onFailure; + } + + public writeEvent(createEvent: () => IDaemonEventEnvelope): void { + this.#enqueue(() => this.#client.writeEventAsync(createEvent())); + } + + public writeLogChunk( + operationId: string, + stream: 'stdout' | 'stderr', + chunk: Uint8Array + ): void { + this.#enqueue(() => this.#client.writeLogChunkAsync(operationId, stream, chunk)); + } + + public async flushAsync(): Promise { + await this.#tail; + if (this.#failure) { + throw this.#failure; + } + } + + #enqueue(writeAsync: () => Promise): void { + this.#tail = this.#tail.then(async () => { + if (this.#failure) { + return; + } + try { + await writeAsync(); + } catch (error) { + this.#failure = error instanceof Error ? error : new Error(String(error)); + this.#onFailure(); + } + }); + } +} + +export class PhasedRequestEventSink implements _IOperationGraphEventSink { + readonly #activeOperationIds: ReadonlySet; + readonly #client: IPhasedRequestClient; + readonly #getNextSequence: () => number; + readonly #observedResults: Map = new Map(); + readonly #rushVersion: string; + readonly #writer: OrderedClientWriter; + + public constructor(options: { + activeOperationIds: ReadonlySet; + client: IPhasedRequestClient; + getNextSequence: () => number; + onWriteFailure: () => void; + rushVersion: string; + }) { + this.#activeOperationIds = options.activeOperationIds; + this.#client = options.client; + this.#getNextSequence = options.getNextSequence; + this.#rushVersion = options.rushVersion; + this.#writer = new OrderedClientWriter(options.client, options.onWriteFailure); + } + + public getObservedResult(operation: Operation): IObservedOperationResult | undefined { + return this.#observedResults.get(operation); + } + + public flushAsync(): Promise { + return this.#writer.flushAsync(); + } + + public onOperationRegistered(operationId: string, silent: boolean): void { + if (this.#activeOperationIds.has(operationId)) { + this.#emitEvent('operationRegistered', { operationId, silent }); + } + } + + public onOperationStatusChanged( + result: IOperationExecutionResult, + previousStatus: OperationStatus + ): void { + const operationId: string = result.operation.name; + if (!this.#activeOperationIds.has(operationId)) { + return; + } + this.#observedResults.set(result.operation, { + executionResult: result, + status: result.status + }); + this.#emitEvent('operationStatusChanged', { + operationId, + previousStatus, + status: result.status + }); + } + + public onOperationHeader(operationId: string, completed: number, total: number): void { + if (this.#activeOperationIds.has(operationId)) { + this.#emitEvent( + 'extension', + { + data: { completedOperations: completed, operationId, totalOperations: total }, + name: RUSHD_OPERATION_HEADER + }, + { required: true } + ); + } + } + + public onOperationChunk(operationId: string, chunk: ITerminalChunk): void { + if (!this.#activeOperationIds.has(operationId)) { + return; + } + const stream: 'stdout' | 'stderr' = + chunk.kind === TerminalChunkKind.Stderr ? 'stderr' : 'stdout'; + this.#writer.writeLogChunk(operationId, stream, TEXT_ENCODER.encode(chunk.text)); + } + + public onOperationStreamClosed(operationId: string): void { + if (this.#activeOperationIds.has(operationId)) { + this.#emitEvent( + 'extension', + { + data: { operationId }, + name: RUSHD_OPERATION_STREAM_CLOSED + }, + { required: true } + ); + } + } + + public onActivity(text: string, options?: _IOperationActivityOptions): void { + const operationId: string | undefined = options?.operationId; + if (operationId !== undefined && !this.#activeOperationIds.has(operationId)) { + return; + } + this.#emitEvent( + 'activityChanged', + { stream: options?.stderr === true ? 'stderr' : 'stdout', text }, + { required: true, scope: operationId === undefined ? undefined : { operationId } } + ); + } + + #emitEvent(type: DaemonEventType, payload: unknown, options?: IEventOptions): void { + this.#writer.writeEvent(() => ({ + eventId: randomUUID(), + payload, + privacy: 'public', + protocolVersion: DAEMON_PROTOCOL_VERSION, + required: options?.required ?? false, + scope: options?.scope, + sequence: this.#getNextSequence(), + sessionId: this.#client.sessionId, + source: { + component: EVENT_SOURCE_COMPONENT, + packageName: EVENT_SOURCE_PACKAGE, + packageVersion: this.#rushVersion + }, + timestamp: new Date().toISOString(), + type + })); + } +} diff --git a/libraries/rush-daemon/src/PhasedRequestRouter.ts b/libraries/rush-daemon/src/PhasedRequestRouter.ts new file mode 100644 index 0000000000..1c4c1049b0 --- /dev/null +++ b/libraries/rush-daemon/src/PhasedRequestRouter.ts @@ -0,0 +1,392 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IOperationExecutionResult, + IOperationGraph, + Operation, + _IOperationGraphEventSink +} from '@microsoft/rush-lib'; +import { OperationStatus } from '@microsoft/rush-lib'; +import type { + IDaemonPhasedEngineShape, + IDaemonPhasedOperationResult, + IDaemonPhasedOperationSelection, + IDaemonPhasedRequest, + IDaemonPhasedRequestResult +} from '@rushstack/rush-daemon-protocol'; + +import { PhasedRequestEventSink } from './PhasedRequestEventSink'; +import { PhasedRequestEventMultiplexer } from './PhasedRequestEventMultiplexer'; +import type { IPhasedRequestClient } from './PhasedRequestClient'; +import { + RequestExclusivityClass, + RequestScheduler, + RequestSchedulerError, + RequestSchedulerErrorCode +} from './RequestScheduler'; +import type { IRequestLease } from './RequestScheduler'; +import type { IWorkspaceEngineShape } from './WorkspaceEngineComponentFactory'; +import type { IWorkspaceSession } from './WorkspaceSession'; + +interface IDualEmitOperationGraph extends IOperationGraph { + eventSink: _IOperationGraphEventSink | undefined; +} + +interface IResolvedSelection { + readonly enabledOperations: ReadonlyArray; + readonly ignoreDependencyOperations: ReadonlyArray; +} + +interface IGraphRoutingState { + readonly multiplexer: PhasedRequestEventMultiplexer; + readonly scheduler: RequestScheduler; +} + +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. + * + * @beta + */ +export class PhasedRequestRouter { + readonly #workspaceSession: IWorkspaceSession; + + public constructor(workspaceSession: IWorkspaceSession) { + this.#workspaceSession = workspaceSession; + } + + /** Validates and executes one resolved phased request against the warm graph. */ + public async executeAsync( + request: IDaemonPhasedRequest, + client: IPhasedRequestClient + ): Promise { + const graph: IDualEmitOperationGraph = getDualEmitGraph(this.#workspaceSession); + const routingState: IGraphRoutingState = getGraphRoutingState(graph); + let lease: IRequestLease; + try { + lease = await routingState.scheduler.acquireAsync({ + abortSignal: client.abortSignal, + exclusivityClass: RequestExclusivityClass.Exclusive + }); + } catch (error) { + if ( + error instanceof RequestSchedulerError && + error.code === RequestSchedulerErrorCode.Aborted + ) { + return createAbortedResult(request.requestId); + } + throw error; + } + + try { + return await this.#executeAdmittedAsync(request, client, graph, routingState); + } finally { + lease.release(); + } + } + + async #executeAdmittedAsync( + request: IDaemonPhasedRequest, + client: IPhasedRequestClient, + graph: IDualEmitOperationGraph, + routingState: IGraphRoutingState + ): Promise { + validateRequestIdentity(request); + validateEngineShape(request.engineShape, this.#workspaceSession.engineShape); + const operationById: ReadonlyMap = indexOperations(graph.operations); + const selection: IResolvedSelection = resolveSelection(request.operationSelection, operationById); + + if (client.abortSignal.aborted) { + return createAbortedResult(request.requestId); + } + if (graph.hasScheduledIteration || graph.status === OperationStatus.Executing) { + throw new Error('The warm workspace operation graph is not idle.'); + } + await this.#workspaceSession.reconcileInvalidationsAsync(); + if (client.abortSignal.aborted) { + return createAbortedResult(request.requestId); + } + + 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) + ); + + 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 }); + + let scheduled: boolean = false; + let executionError: unknown; + const iterationCleanupErrors: unknown[] = []; + 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; + } + } 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 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); + } + await abortTail; + cleanupErrors.push(...abortErrors.slice(observedAbortErrorCount)); + throwCombinedErrors(executionError, cleanupErrors); + + return { + aborted: wasAborted || client.abortSignal.aborted, + operationResults: collectOperationResults(activeOperations, graph, requestSink), + requestId: request.requestId, + scheduled + }; + } +} + +function getDualEmitGraph(workspaceSession: IWorkspaceSession): IDualEmitOperationGraph { + const graph: IOperationGraph | undefined = workspaceSession.operationGraph; + if (!graph) { + throw new Error('The workspace session does not provide a reusable operation graph.'); + } + if (!('eventSink' in graph)) { + throw new Error('The workspace operation graph does not support Rush dual-emit events.'); + } + return graph as IDualEmitOperationGraph; +} + +function getGraphEventSink(graph: IDualEmitOperationGraph): _IOperationGraphEventSink | undefined { + return graph.eventSink; +} + +function setGraphEventSink( + graph: IDualEmitOperationGraph, + eventSink: _IOperationGraphEventSink | undefined +): void { + graph.eventSink = eventSink; +} + +function setPauseNextIteration(graph: IOperationGraph, pauseNextIteration: boolean): void { + graph.pauseNextIteration = pauseNextIteration; +} + +function getGraphRoutingState(graph: IDualEmitOperationGraph): IGraphRoutingState { + let state: IGraphRoutingState | undefined = ROUTING_STATE_BY_GRAPH.get(graph); + if (!state) { + const multiplexer: PhasedRequestEventMultiplexer = new PhasedRequestEventMultiplexer( + getGraphEventSink(graph) + ); + state = { multiplexer, scheduler: new RequestScheduler() }; + ROUTING_STATE_BY_GRAPH.set(graph, state); + setGraphEventSink(graph, multiplexer); + } else if (getGraphEventSink(graph) !== state.multiplexer) { + throw new Error('The workspace operation graph event sink changed after routing began.'); + } + return state; +} + +function validateRequestIdentity(request: IDaemonPhasedRequest): void { + validateNonemptyName(request.requestId, 'request id'); + validateNonemptyName(request.commandName, 'command name'); +} + +function validateNonemptyName(value: string, kind: string): void { + if (value.length === 0 || value.trim() !== value) { + throw new Error(`Invalid phased request ${kind}: "${value}".`); + } +} + +function validateEngineShape( + requestShape: IDaemonPhasedEngineShape, + workspaceShape: IWorkspaceEngineShape | undefined +): void { + if (!workspaceShape) { + throw new Error('The workspace session does not declare a reusable engine shape.'); + } + validateNameSet(requestShape.phaseNames, workspaceShape.phaseNames, 'phase'); + validateNameSet(requestShape.pluginNames, workspaceShape.pluginNames, 'plugin'); +} + +function validateNameSet( + requestedNames: ReadonlyArray, + workspaceNames: ReadonlyArray, + kind: string +): void { + const requested: Set = new Set(requestedNames); + if ( + requested.size !== requestedNames.length || + requested.size !== workspaceNames.length || + workspaceNames.some((name: string) => !requested.has(name)) + ) { + throw new Error(`The phased request ${kind} shape does not match the warm workspace engine.`); + } +} + +function indexOperations(operations: ReadonlySet): ReadonlyMap { + const operationById: Map = new Map(); + for (const operation of operations) { + const operationId: string = operation.name; + if (operationById.has(operationId)) { + throw new Error(`The workspace graph contains duplicate operation id "${operationId}".`); + } + operationById.set(operationId, operation); + } + return operationById; +} + +function resolveSelection( + requestedSelection: ReadonlyArray, + operationById: ReadonlyMap +): IResolvedSelection { + if (requestedSelection.length === 0) { + throw new Error('A phased request must select at least one operation.'); + } + const selectedIds: Set = new Set(); + const enabledOperations: Operation[] = []; + const ignoreDependencyOperations: Operation[] = []; + for (const selection of requestedSelection) { + validateNonemptyName(selection.operationId, 'operation id'); + if (selectedIds.has(selection.operationId)) { + throw new Error(`Duplicate phased request operation id "${selection.operationId}".`); + } + selectedIds.add(selection.operationId); + const operation: Operation | undefined = operationById.get(selection.operationId); + if (!operation) { + throw new Error(`Unknown phased request operation id "${selection.operationId}".`); + } + addSelectedOperation(selection.enabledState, operation, enabledOperations, ignoreDependencyOperations); + } + return { enabledOperations, ignoreDependencyOperations }; +} + +function addSelectedOperation( + enabledState: unknown, + operation: Operation, + enabledOperations: Operation[], + ignoreDependencyOperations: Operation[] +): void { + if (enabledState === true) { + enabledOperations.push(operation); + } else if (enabledState === 'ignore-dependency-changes') { + ignoreDependencyOperations.push(operation); + } else { + throw new Error(`Invalid phased request enabled state: "${String(enabledState)}".`); + } +} + +function applySelection(graph: IOperationGraph, selection: IResolvedSelection): void { + graph.setEnabledStates(graph.operations, false, 'unsafe'); + graph.setEnabledStates( + selection.ignoreDependencyOperations, + 'ignore-dependency-changes', + 'safe' + ); + graph.setEnabledStates(selection.enabledOperations, true, 'safe'); + graph.setEnabledStates( + selection.ignoreDependencyOperations, + 'ignore-dependency-changes', + 'unsafe' + ); +} + +function collectOperationResults( + activeOperations: ReadonlyArray, + graph: IOperationGraph, + requestSink: PhasedRequestEventSink +): ReadonlyArray { + const results: IDaemonPhasedOperationResult[] = []; + 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; + if (status === undefined) { + continue; + } + const errorMessage: string | undefined = observed + ? observed.executionResult.error?.message + : retained?.error?.message; + results.push({ operationId: operation.name, status, errorMessage }); + } + return results; +} + +function compareOperations(left: Operation, right: Operation): number { + return left.name.localeCompare(right.name); +} + +function createAbortedResult(requestId: string): IDaemonPhasedRequestResult { + return { aborted: true, operationResults: [], requestId, scheduled: false }; +} + +function throwCombinedErrors(executionError: unknown, cleanupErrors: unknown[]): void { + if (executionError !== undefined && cleanupErrors.length > 0) { + throw new AggregateError( + [executionError, ...cleanupErrors], + 'The phased request failed and could not clean up its client subscription.' + ); + } + if (executionError !== undefined) { + throw executionError; + } + if (cleanupErrors.length === 1) { + throw cleanupErrors[0]; + } + if (cleanupErrors.length > 1) { + throw new AggregateError(cleanupErrors, 'Failed to clean up the phased request client subscription.'); + } +} diff --git a/libraries/rush-daemon/src/index.ts b/libraries/rush-daemon/src/index.ts index 2de9d94142..a49fb7bad6 100644 --- a/libraries/rush-daemon/src/index.ts +++ b/libraries/rush-daemon/src/index.ts @@ -42,3 +42,5 @@ export { WorkspaceInvalidationTracker, type IWorkspaceInvalidationSnapshot } from './WorkspaceInvalidationTracker'; +export { type IPhasedRequestClient } from './PhasedRequestClient'; +export { PhasedRequestRouter } from './PhasedRequestRouter'; diff --git a/libraries/rush-daemon/src/test/PhasedRequestEventSink.test.ts b/libraries/rush-daemon/src/test/PhasedRequestEventSink.test.ts new file mode 100644 index 0000000000..ac5bfdfe06 --- /dev/null +++ b/libraries/rush-daemon/src/test/PhasedRequestEventSink.test.ts @@ -0,0 +1,46 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol'; + +import { PhasedRequestEventSink } from '../PhasedRequestEventSink'; +import { TestPhasedRequestClient } from './PhasedRequestRouterTestUtilities'; + +const ACTIVE_OPERATION: string = 'project-a (_phase:test)'; +const OTHER_OPERATION: string = 'project-b (_phase:test)'; + +function createSink(client: TestPhasedRequestClient): PhasedRequestEventSink { + return new PhasedRequestEventSink({ + activeOperationIds: new Set([ACTIVE_OPERATION]), + client, + getNextSequence: () => client.getNextEventSequence(), + onWriteFailure: () => undefined, + rushVersion: '5.178.1' + }); +} + +it('forwards unscoped and active activity while filtering other operation activity', async () => { + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + const sink: PhasedRequestEventSink = createSink(client); + sink.onActivity('request summary'); + sink.onActivity('active detail', { operationId: ACTIVE_OPERATION }); + sink.onActivity('other detail', { operationId: OTHER_OPERATION }); + + await sink.flushAsync(); + + const activities: IDaemonEventEnvelope[] = client.writes + .map(({ event }) => event) + .filter( + (event: IDaemonEventEnvelope | undefined): event is IDaemonEventEnvelope => + event?.type === 'activityChanged' + ); + expect(activities.map(({ payload }) => payload)).toEqual([ + { stream: 'stdout', text: 'request summary' }, + { stream: 'stdout', text: 'active detail' } + ]); + expect(activities.map(({ scope }) => scope)).toEqual([ + undefined, + { operationId: ACTIVE_OPERATION } + ]); + expect(activities.every(({ required }) => required)).toBe(true); +}); diff --git a/libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts b/libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts new file mode 100644 index 0000000000..e5f2bf9471 --- /dev/null +++ b/libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts @@ -0,0 +1,570 @@ +// 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 +} from '@rushstack/rush-daemon-protocol'; +import { + RUSHD_OPERATION_HEADER, + RUSHD_OPERATION_STREAM_CLOSED +} 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 { + ITestClientWrite, + 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)'; + +function createRequest( + operationSelection: ReadonlyArray +): IDaemonPhasedRequest { + return { + commandName: 'build', + engineShape: TEST_ENGINE_SHAPE, + operationSelection, + requestId: 'request-1' + }; +} + +function select(operationId: string): IDaemonPhasedOperationSelection { + return { enabledState: true, operationId }; +} + +function selectRuntimeValue( + operationId: string, + enabledState: unknown +): IDaemonPhasedOperationSelection { + return { enabledState, operationId } as unknown as IDaemonPhasedOperationSelection; +} + +function createThreeOperationFixture(options?: { + actionAAsync?: (terminal: ITerminal) => Promise; + 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)] + ]), + [[OPERATION_B, OPERATION_A]] + ); +} + +function getEventOperationId(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; + if (typeof data !== 'object' || data === null) { + return undefined; + } + const nestedOperationId: unknown = (data as { operationId?: unknown }).operationId; + return typeof nestedOperationId === 'string' ? nestedOperationId : undefined; +} + +describe(PhasedRequestRouter.name, () => { + it('rejects invalid selections and an engine-shape mismatch before scheduling', async () => { + const fixture: ITestRoutingFixture = createThreeOperationFixture(); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + + await expect(router.executeAsync(createRequest([]), client)).rejects.toThrow( + 'must select at least one operation' + ); + await expect( + router.executeAsync(createRequest([select('unknown operation')]), client) + ).rejects.toThrow('Unknown phased request operation id'); + await expect( + router.executeAsync(createRequest([select(OPERATION_A), select(OPERATION_A)]), client) + ).rejects.toThrow('Duplicate phased request operation id'); + for (const enabledState of [false, 'invalid-state']) { + await expect( + router.executeAsync( + createRequest([selectRuntimeValue(OPERATION_A, enabledState)]), + client + ) + ).rejects.toThrow(`Invalid phased request enabled state: "${String(enabledState)}"`); + } + await expect( + router.executeAsync( + { ...createRequest([select(OPERATION_A)]), engineShape: { phaseNames: ['other'], pluginNames: [] } }, + client + ) + ).rejects.toThrow('phase shape does not match'); + expect(scheduleSpy).not.toHaveBeenCalled(); + }); + + it('accepts both enabled states declared by the protocol', async () => { + const trueFixture: ITestRoutingFixture = createThreeOperationFixture(); + await new PhasedRequestRouter(trueFixture.session).executeAsync( + createRequest([select(OPERATION_A)]), + new TestPhasedRequestClient() + ); + + const ignoredDependencyFixture: ITestRoutingFixture = createThreeOperationFixture(); + await new PhasedRequestRouter(ignoredDependencyFixture.session).executeAsync( + createRequest([ + { + enabledState: 'ignore-dependency-changes', + operationId: OPERATION_A + } + ]), + new TestPhasedRequestClient() + ); + + expect(trueFixture.operations.get(OPERATION_A)?.enabled).toBe(true); + expect(ignoredDependencyFixture.operations.get(OPERATION_A)?.enabled).toBe( + 'ignore-dependency-changes' + ); + + const mixedFixture: ITestRoutingFixture = createThreeOperationFixture(); + await new PhasedRequestRouter(mixedFixture.session).executeAsync( + createRequest([ + { + enabledState: 'ignore-dependency-changes', + operationId: OPERATION_A + }, + select(OPERATION_B) + ]), + new TestPhasedRequestClient() + ); + + expect(mixedFixture.operations.get(OPERATION_A)?.enabled).toBe( + 'ignore-dependency-changes' + ); + expect(mixedFixture.operations.get(OPERATION_B)?.enabled).toBe(true); + }); + + it('reconciles invalidations, applies the safe dependency closure, and runs one iteration', async () => { + const order: string[] = []; + const fixture: ITestRoutingFixture = createThreeOperationFixture({ + actionAAsync: async (): Promise => { + order.push('run'); + } + }); + fixture.session.onReconcileAsync = async (): Promise => { + order.push('reconcile'); + }; + fixture.graph.hooks.onIterationScheduled.tap('test', () => { + order.push('schedule'); + }); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + const executeSpy: jest.SpyInstance = jest.spyOn( + fixture.graph, + 'executeScheduledIterationAsync' + ); + + const result = await new PhasedRequestRouter(fixture.session).executeAsync( + createRequest([select(OPERATION_B)]), + new TestPhasedRequestClient() + ); + + expect(order).toEqual(['reconcile', 'schedule', 'run']); + expect(scheduleSpy).toHaveBeenCalledTimes(1); + expect(executeSpy).toHaveBeenCalledTimes(1); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(1); + expect(fixture.runners.get(OPERATION_B)?.runCount).toBe(1); + expect(fixture.runners.get(OPERATION_C)?.runCount).toBe(0); + expect(result.operationResults.map(({ operationId }) => operationId)).toEqual([ + OPERATION_A, + OPERATION_B + ]); + expect(result.scheduled).toBe(true); + }); + + it('forwards only enabled operations with ordered client backpressure', async () => { + const fixture: ITestRoutingFixture = createThreeOperationFixture({ + actionAAsync: async (terminal: ITerminal): Promise => { + terminal.writeLine('stdout-a'); + terminal.writeErrorLine('stderr-a', { doNotOverrideSgrCodes: true }); + } + }); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + let concurrentWrites: number = 0; + let maximumConcurrentWrites: number = 0; + client.onWriteAsync = async (): Promise => { + concurrentWrites++; + maximumConcurrentWrites = Math.max(maximumConcurrentWrites, concurrentWrites); + await new Promise((resolve) => setImmediate(resolve)); + concurrentWrites--; + }; + + await new PhasedRequestRouter(fixture.session).executeAsync( + createRequest([select(OPERATION_A)]), + client + ); + + expect(maximumConcurrentWrites).toBe(1); + const logWrites: ITestClientWrite[] = client.writes.filter( + (write: ITestClientWrite) => write.text !== undefined + ); + expect(logWrites.map(({ operationId }) => operationId)).toEqual([ + OPERATION_A, + OPERATION_A + ]); + expect(logWrites.map(({ stream }) => stream)).toEqual(['stdout', 'stderr']); + expect(logWrites.map(({ text }) => text)).toEqual(['stdout-a\n', 'stderr-a\n']); + const eventOperationIds: string[] = client.writes + .map((write: ITestClientWrite) => write.event) + .filter((event: IDaemonEventEnvelope | undefined): event is IDaemonEventEnvelope => !!event) + .map(getEventOperationId) + .filter((operationId: string | undefined): operationId is string => !!operationId); + expect(new Set(eventOperationIds)).toEqual(new Set([OPERATION_A])); + const streamClosedEvent: IDaemonEventEnvelope | undefined = client.writes + .map((write: ITestClientWrite) => write.event) + .find( + (event: IDaemonEventEnvelope | undefined) => + (event?.payload as { name?: unknown } | undefined)?.name === + RUSHD_OPERATION_STREAM_CLOSED + ); + expect(streamClosedEvent?.required).toBe(true); + }); + + it('allocates event sequences when queued writes are invoked', async () => { + let releaseFirstEvent: (() => void) | undefined; + let markFirstEventStarted: (() => void) | undefined; + const firstEventStarted: Promise = new Promise((resolve) => { + markFirstEventStarted = resolve; + }); + const fixture: ITestRoutingFixture = createThreeOperationFixture(); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + let hasBlockedEvent: boolean = false; + client.onWriteAsync = async (write: ITestClientWrite): Promise => { + if (write.event && !hasBlockedEvent) { + hasBlockedEvent = true; + markFirstEventStarted?.(); + await new Promise((resolve) => { + releaseFirstEvent = resolve; + }); + } + }; + + const requestPromise = new PhasedRequestRouter(fixture.session).executeAsync( + createRequest([select(OPERATION_A)]), + client + ); + await firstEventStarted; + const interleavedSequence: number = client.getNextEventSequence(); + releaseFirstEvent?.(); + await requestPromise; + + const routedSequences: number[] = client.writes + .map(({ event }) => event?.sequence) + .filter((sequence: number | undefined): sequence is number => sequence !== undefined); + expect(routedSequences[0]).toBe(1); + expect(interleavedSequence).toBe(2); + expect(routedSequences.slice(1).every((sequence) => sequence > interleavedSequence)).toBe(true); + }); + + it('returns client-scoped failures without converting them to routing errors', async () => { + const fixture: ITestRoutingFixture = createThreeOperationFixture({ + statusA: OperationStatus.Failure + }); + + const result = await new PhasedRequestRouter(fixture.session).executeAsync( + createRequest([select(OPERATION_A)]), + new TestPhasedRequestClient() + ); + + expect(result.operationResults).toEqual([ + { errorMessage: undefined, operationId: OPERATION_A, status: OperationStatus.Failure } + ]); + }); + + it('aborts a cancelled iteration, restores subscriptions, and keeps runners reusable', async () => { + let releaseOperationA: (() => void) | undefined; + let markOperationAStarted: (() => void) | undefined; + const operationAStarted: Promise = new Promise((resolve) => { + markOperationAStarted = resolve; + }); + const fixture: ITestRoutingFixture = createThreeOperationFixture({ + actionAAsync: async (): Promise => { + markOperationAStarted?.(); + await new Promise((resolve) => { + releaseOperationA = resolve; + }); + } + }); + let workspaceStatusEventCount: number = 0; + const previousSink = { + onOperationStatusChanged: (): void => { + workspaceStatusEventCount++; + } + }; + fixture.graph.eventSink = previousSink; + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const requestPromise = router.executeAsync( + createRequest([select(OPERATION_B)]), + client + ); + await operationAStarted; + client.abortController.abort(); + releaseOperationA?.(); + + const result = await requestPromise; + + expect(result.aborted).toBe(true); + expect( + result.operationResults.find(({ operationId }) => operationId === OPERATION_B)?.status + ).toBe(OperationStatus.Aborted); + expect(fixture.graph.pauseNextIteration).toBe(false); + expect(fixture.runners.get(OPERATION_A)?.closeCount).toBe(0); + expect(fixture.runners.get(OPERATION_B)?.closeCount).toBe(0); + const completedClientWriteCount: number = client.writes.length; + fixture.graph.invalidateOperations(undefined, 'after request'); + expect(client.writes).toHaveLength(completedClientWriteCount); + expect(workspaceStatusEventCount).toBeGreaterThan(0); + + const followUp = await router.executeAsync( + createRequest([select(OPERATION_C)]), + new TestPhasedRequestClient() + ); + expect(followUp.operationResults[0]?.status).toBe(OperationStatus.Success); + }); + + it('does not combine an observed result with an error retained from a prior iteration', async () => { + let invocation: number = 0; + let releaseOperationA: (() => void) | undefined; + let markOperationAStarted: (() => void) | undefined; + const operationAStarted: Promise = new Promise((resolve) => { + markOperationAStarted = resolve; + }); + const fixture: ITestRoutingFixture = createThreeOperationFixture({ + actionAAsync: async (): Promise => { + if (invocation++ === 0) { + throw new Error('first iteration failure'); + } + markOperationAStarted?.(); + await new Promise((resolve) => { + releaseOperationA = resolve; + }); + } + }); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const first = await router.executeAsync( + createRequest([select(OPERATION_B)]), + new TestPhasedRequestClient() + ); + expect( + first.operationResults.find(({ operationId }) => operationId === OPERATION_A)?.errorMessage + ).toBe('first iteration failure'); + + const secondClient: TestPhasedRequestClient = new TestPhasedRequestClient(); + const secondPromise = router.executeAsync( + { ...createRequest([select(OPERATION_B)]), requestId: 'request-2' }, + secondClient + ); + await operationAStarted; + secondClient.abortController.abort(); + releaseOperationA?.(); + const second = await secondPromise; + + expect( + second.operationResults.find(({ operationId }) => operationId === OPERATION_A)?.errorMessage + ).toBeUndefined(); + }); + + it('aborts and unsubscribes when a disconnected client rejects a write', async () => { + let releaseOperationA: (() => void) | undefined; + const fixture: ITestRoutingFixture = createThreeOperationFixture({ + actionAAsync: async (terminal: ITerminal): Promise => { + terminal.writeLine('disconnect'); + await new Promise((resolve) => { + releaseOperationA = resolve; + }); + } + }); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + client.onWriteAsync = async (write: ITestClientWrite): Promise => { + if (write.text !== undefined) { + releaseOperationA?.(); + throw new Error('client disconnected'); + } + }; + + await expect( + new PhasedRequestRouter(fixture.session).executeAsync( + createRequest([select(OPERATION_B)]), + client + ) + ).rejects.toThrow('client disconnected'); + expect(fixture.graph.pauseNextIteration).toBe(false); + expect(fixture.runners.get(OPERATION_A)?.closeCount).toBe(0); + }); + + it('re-aborts after an early write failure crosses the schedule boundary', async () => { + const fixture: ITestRoutingFixture = createThreeOperationFixture(); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + client.onWriteAsync = async (): Promise => { + throw new Error('client disconnected before execution'); + }; + + await expect( + new PhasedRequestRouter(fixture.session).executeAsync( + createRequest([select(OPERATION_B)]), + client + ) + ).rejects.toThrow('client disconnected before execution'); + expect(fixture.runners.get(OPERATION_B)?.runCount).toBe(0); + expect(fixture.graph.hasScheduledIteration).toBe(false); + }); + + it('serializes independent router instances that share one warm graph', async () => { + let releaseOperationA: (() => void) | undefined; + let markOperationAStarted: (() => void) | undefined; + const operationAStarted: Promise = new Promise((resolve) => { + markOperationAStarted = resolve; + }); + const fixture: ITestRoutingFixture = createThreeOperationFixture({ + actionAAsync: async (): Promise => { + markOperationAStarted?.(); + await new Promise((resolve) => { + releaseOperationA = resolve; + }); + } + }); + const firstPromise = new PhasedRequestRouter(fixture.session).executeAsync( + createRequest([select(OPERATION_A)]), + new TestPhasedRequestClient() + ); + await operationAStarted; + const secondPromise = new PhasedRequestRouter(fixture.session).executeAsync( + { ...createRequest([select(OPERATION_C)]), requestId: 'request-2' }, + new TestPhasedRequestClient() + ); + await Promise.resolve(); + expect(fixture.runners.get(OPERATION_C)?.runCount).toBe(0); + releaseOperationA?.(); + + await Promise.all([firstPromise, secondPromise]); + expect(fixture.runners.get(OPERATION_C)?.runCount).toBe(1); + }); + + it('drains and aborts a scheduled iteration when a scheduling hook fails', async () => { + const fixture: ITestRoutingFixture = createThreeOperationFixture(); + fixture.graph.hooks.onIterationScheduled.tap('throwing test hook', () => { + throw new Error('scheduling hook failed'); + }); + + await expect( + new PhasedRequestRouter(fixture.session).executeAsync( + createRequest([select(OPERATION_A)]), + new TestPhasedRequestClient() + ) + ).rejects.toThrow('scheduling hook failed'); + expect(fixture.graph.hasScheduledIteration).toBe(false); + expect(fixture.graph.status).not.toBe(OperationStatus.Executing); + const completedRunCount: number = fixture.runners.get(OPERATION_A)?.runCount ?? 0; + await new Promise((resolve) => setImmediate(resolve)); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(completedRunCount); + }); + + it('returns retained results when real graph hooks collapse a repeated warm request to no work', async () => { + const fixture: ITestRoutingFixture = createThreeOperationFixture(); + let iteration: number = 0; + fixture.graph.hooks.configureIteration.tap('warm no-op', (records, previousResults) => { + if (iteration++ === 0) { + return; + } + for (const record of records.values()) { + if (previousResults.has(record.operation)) { + record.enabled = false; + } + } + }); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const sequenceState: { next: number } = { next: 1 }; + const firstClient: TestPhasedRequestClient = new TestPhasedRequestClient(sequenceState); + const secondClient: TestPhasedRequestClient = new TestPhasedRequestClient(sequenceState); + + const first = await router.executeAsync( + createRequest([select(OPERATION_A)]), + firstClient + ); + const second = await router.executeAsync( + { ...createRequest([select(OPERATION_A)]), requestId: 'request-2' }, + secondClient + ); + + expect(first.scheduled).toBe(true); + expect(second.scheduled).toBe(false); + expect(second.operationResults).toEqual(first.operationResults); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(1); + const firstSequences: number[] = firstClient.writes + .map(({ event }) => event?.sequence) + .filter((sequence: number | undefined): sequence is number => sequence !== undefined); + const secondSequences: number[] = secondClient.writes + .map(({ event }) => event?.sequence) + .filter((sequence: number | undefined): sequence is number => sequence !== undefined); + expect(firstSequences.length).toBeGreaterThan(0); + expect(secondSequences[0]).toBeGreaterThan(firstSequences[firstSequences.length - 1]); + }); + + it('uses authoritative header totals after a partial warm iteration', async () => { + const fixture: ITestRoutingFixture = createThreeOperationFixture(); + let iteration: number = 0; + fixture.graph.hooks.configureIteration.tap('partial warm iteration', (records, previousResults) => { + if (iteration++ === 0) { + return; + } + for (const record of records.values()) { + if (record.operation.name === OPERATION_A && previousResults.has(record.operation)) { + record.enabled = false; + } + } + }); + const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + await router.executeAsync(createRequest([select(OPERATION_B)]), new TestPhasedRequestClient()); + const client: TestPhasedRequestClient = new TestPhasedRequestClient(); + + await router.executeAsync( + { ...createRequest([select(OPERATION_B)]), requestId: 'request-2' }, + client + ); + + const headers: IDaemonEventEnvelope[] = client.writes + .map(({ event }) => event) + .filter( + (event: IDaemonEventEnvelope | undefined): event is IDaemonEventEnvelope => + (event?.payload as { name?: unknown } | undefined)?.name === RUSHD_OPERATION_HEADER + ); + expect(headers).toHaveLength(1); + expect(headers[0]?.required).toBe(true); + expect(headers[0]?.payload).toEqual({ + data: { completedOperations: 1, operationId: OPERATION_B, totalOperations: 1 }, + name: RUSHD_OPERATION_HEADER + }); + }); +}); diff --git a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts new file mode 100644 index 0000000000..bde8d5885c --- /dev/null +++ b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts @@ -0,0 +1,233 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { MockWritable } from '@rushstack/terminal'; +import type { ITerminal } from '@rushstack/terminal'; +import type { + IInputsSnapshot, + IOperationGraph, + IOperationRunner, + IOperationRunnerContext, + IPhase, + RushConfiguration, + RushSession +} from '@microsoft/rush-lib'; +import { Operation, OperationStatus } from '@microsoft/rush-lib'; +import { OperationGraph } from '@microsoft/rush-lib/lib/logic/operations/OperationGraph'; +import type { IOperationGraphOptions } from '@microsoft/rush-lib/lib/logic/operations/OperationGraph'; +import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol'; + +import type { IPhasedRequestClient } from '../PhasedRequestClient'; +import type { + IWorkspaceEngineShape, + IWorkspaceInvalidationReconciliation +} from '../WorkspaceEngineComponentFactory'; +import type { + IWorkspaceSession, + IWorkspaceSessionMetadata +} from '../WorkspaceSession'; +import { WorkspaceInvalidationTracker } from '../WorkspaceInvalidationTracker'; +import { TEST_RUSH_CONFIGURATION, TEST_REPO_ROOT } from './TestWorkspaceSession'; + +export const TEST_ENGINE_SHAPE: IWorkspaceEngineShape = { + phaseNames: ['_phase:test'], + pluginNames: ['test-plugin'] +}; + +const TEST_PHASE: IPhase = { + allowWarningsOnSuccess: false, + associatedParameters: new Set(), + dependencies: { self: new Set(), upstream: new Set() }, + isSynthetic: false, + logFilenameIdentifier: '_phase_test', + missingScriptBehavior: 'silent', + name: TEST_ENGINE_SHAPE.phaseNames[0] +}; + +export interface ITestClientWrite { + readonly event?: IDaemonEventEnvelope; + readonly operationId?: string; + readonly stream?: 'stdout' | 'stderr'; + readonly text?: string; +} + +export class TestPhasedRequestClient implements IPhasedRequestClient { + public readonly abortController: AbortController = new AbortController(); + public readonly sessionId: string = 'test-session'; + public readonly writes: ITestClientWrite[] = []; + public onWriteAsync: ((write: ITestClientWrite) => Promise) | undefined; + readonly #sequenceState: { next: number }; + + public constructor(sequenceState: { next: number } = { next: 1 }) { + this.#sequenceState = sequenceState; + } + + public get abortSignal(): AbortSignal { + return this.abortController.signal; + } + + public getNextEventSequence(): number { + const sequence: number = this.#sequenceState.next; + this.#sequenceState.next = sequence + 1; + return sequence; + } + + public async writeEventAsync(event: IDaemonEventEnvelope): Promise { + const write: ITestClientWrite = { event }; + await this.onWriteAsync?.(write); + this.writes.push(write); + } + + public async writeLogChunkAsync( + operationId: string, + stream: 'stdout' | 'stderr', + chunk: Uint8Array + ): Promise { + const write: ITestClientWrite = { + operationId, + stream, + text: new TextDecoder().decode(chunk) + }; + await this.onWriteAsync?.(write); + this.writes.push(write); + } +} + +export class TestOperationRunner implements IOperationRunner { + public readonly cacheable: boolean = false; + public readonly reportTiming: boolean = true; + public readonly silent: boolean = false; + public readonly warningsAreAllowed: boolean = false; + public closeCount: number = 0; + public runCount: number = 0; + + readonly #actionAsync: ((terminal: ITerminal) => Promise) | undefined; + readonly #status: OperationStatus; + public readonly name: string; + + public constructor( + name: string, + status: OperationStatus = OperationStatus.Success, + actionAsync?: (terminal: ITerminal) => Promise + ) { + this.name = name; + this.#status = status; + this.#actionAsync = actionAsync; + } + + public closeAsync(): Promise { + this.closeCount++; + return Promise.resolve(); + } + + public executeAsync(context: IOperationRunnerContext): Promise { + this.runCount++; + return context.runWithTerminalAsync( + async (terminal: ITerminal): Promise => { + await this.#actionAsync?.(terminal); + return this.#status; + }, + { createLogFile: false, logFileSuffix: '' } + ); + } + + public getConfigHash(): string { + return this.name; + } +} + +export interface ITestRoutingFixture { + readonly graph: OperationGraph; + readonly operations: ReadonlyMap; + readonly runners: ReadonlyMap; + readonly session: TestRoutingWorkspaceSession; +} + +export class TestRoutingWorkspaceSession implements IWorkspaceSession { + public readonly engineShape: IWorkspaceEngineShape = TEST_ENGINE_SHAPE; + public readonly inputsSnapshot: IInputsSnapshot | undefined = undefined; + public readonly invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + public readonly metadata: IWorkspaceSessionMetadata = { + projectCount: 3, + projectNames: ['project-a', 'project-b', 'project-c'], + repoRoot: TEST_REPO_ROOT, + rushJsonFile: TEST_RUSH_CONFIGURATION.rushJsonFile, + rushVersion: '5.178.1' + }; + public readonly rushConfiguration: RushConfiguration = TEST_RUSH_CONFIGURATION; + public readonly rushSession: RushSession | undefined = undefined; + public readonly operationGraph: IOperationGraph; + public onReconcileAsync: (() => Promise) | undefined; + + public constructor(operationGraph: IOperationGraph) { + this.operationGraph = operationGraph; + } + + public async reconcileInvalidationsAsync(): Promise< + IWorkspaceInvalidationReconciliation | undefined + > { + await this.onReconcileAsync?.(); + return undefined; + } + + public async [Symbol.asyncDispose](): Promise { + this.operationGraph.abortController.abort(); + await this.operationGraph.abortCurrentIterationAsync(); + await this.operationGraph.closeRunnersAsync(); + } +} + +export function createRoutingFixture( + runnerById: ReadonlyMap, + dependencies: ReadonlyArray = [] +): ITestRoutingFixture { + const operations: Map = new Map(); + const runners: Map = new Map(runnerById); + let projectIndex: number = 0; + for (const [operationId, runner] of runners) { + const project = TEST_RUSH_CONFIGURATION.projects[projectIndex++]; + if (!project) { + throw new Error('The test Rush configuration does not have enough projects.'); + } + operations.set( + operationId, + new Operation({ + logFilenameIdentifier: operationId, + phase: TEST_PHASE, + project, + runner + }) + ); + } + for (const [consumerId, dependencyId] of dependencies) { + const consumer: Operation | undefined = operations.get(consumerId); + const dependency: Operation | undefined = operations.get(dependencyId); + if (!consumer || !dependency) { + throw new Error('The test dependency references an unknown operation.'); + } + consumer.addDependency(dependency); + } + + const graphOptions: IOperationGraphOptions = { + abortController: new AbortController(), + allowOversubscription: true, + debugMode: false, + destinations: [new MockWritable()], + parallelism: 1, + pauseNextIteration: false, + quietMode: false + }; + // The package's bundled public declarations and deep-import declarations describe the same runtime classes, + // but TypeScript assigns them distinct recursive identities. + const graph: OperationGraph = new OperationGraph( + new Set(operations.values()) as unknown as ConstructorParameters[0], + graphOptions + ); + const publicGraph: IOperationGraph = graph as unknown as IOperationGraph; + return { + graph, + operations, + runners, + session: new TestRoutingWorkspaceSession(publicGraph) + }; +} diff --git a/libraries/rush-terminal-renderer/src/HostEventRouter.ts b/libraries/rush-terminal-renderer/src/HostEventRouter.ts index a5a81ba167..3f26e146eb 100644 --- a/libraries/rush-terminal-renderer/src/HostEventRouter.ts +++ b/libraries/rush-terminal-renderer/src/HostEventRouter.ts @@ -5,8 +5,10 @@ import { type DaemonVerbosity, type IDaemonEventEnvelope, type IDaemonExtensionEventPayload, + type IDaemonOperationHeaderPayload, type IDaemonOperationRegisteredPayload, type IDaemonOperationStreamClosedPayload, + RUSHD_OPERATION_HEADER, RUSHD_OPERATION_STREAM_CLOSED, shouldSerializeDaemonEvent } from '@rushstack/rush-daemon-protocol'; @@ -20,11 +22,7 @@ function readScopeOperationId(envelope: IDaemonEventEnvelope): string | undefine return scope === undefined ? undefined : scope.operationId; } -/** - * Routes decoded event envelopes between the collator (stream-affecting and - * operation-scoped events) and the verbosity-filtered renderer. - * @internal - */ +/** Routes decoded events between the operation collator and renderer. @internal */ export class HostEventRouter { private readonly _streams: OperationStreamRegistry; private readonly _renderer: IDaemonRenderer; @@ -40,7 +38,6 @@ export class HostEventRouter { this._verbosity = verbosity; } - /** Routes one decoded `0x05` event envelope. */ public routeEvent(envelope: IDaemonEventEnvelope): void { this._trackOperationLifecycle(envelope); if (this._routeScopedActivity(envelope)) { @@ -67,6 +64,10 @@ export class HostEventRouter { } private _trackExtension(payload: IDaemonExtensionEventPayload): void { + if (payload.name === RUSHD_OPERATION_HEADER) { + this._streams.setOperationHeader(payload.data as IDaemonOperationHeaderPayload); + return; + } if (payload.name === RUSHD_OPERATION_STREAM_CLOSED) { const data: IDaemonOperationStreamClosedPayload = payload.data as IDaemonOperationStreamClosedPayload; @@ -74,9 +75,6 @@ export class HostEventRouter { } } - // Operation-scoped activity lines are part of the operation's output block - // (legacy writes them to the operation's collated stream, bypassing the - // quiet-mode stdout discard), so they route to the collator, not the renderer. private _routeScopedActivity(envelope: IDaemonEventEnvelope): boolean { const operationId: string | undefined = readScopeOperationId(envelope); if (envelope.type !== 'activityChanged' || operationId === undefined) { diff --git a/libraries/rush-terminal-renderer/src/OperationHeaderTracker.ts b/libraries/rush-terminal-renderer/src/OperationHeaderTracker.ts new file mode 100644 index 0000000000..1ee286ec64 --- /dev/null +++ b/libraries/rush-terminal-renderer/src/OperationHeaderTracker.ts @@ -0,0 +1,38 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonOperationHeaderPayload } from '@rushstack/rush-daemon-protocol'; + +const INITIAL_OPERATION_COUNT: number = 0; +const OPERATION_COUNT_INCREMENT: number = 1; + +export class OperationHeaderTracker { + private readonly _headerByOperation: Map = new Map(); + private _completedOperations: number = INITIAL_OPERATION_COUNT; + private _totalOperations: number = INITIAL_OPERATION_COUNT; + + public registerOperation(): void { + this._totalOperations += OPERATION_COUNT_INCREMENT; + } + + public setOperationHeader(header: IDaemonOperationHeaderPayload): void { + this._headerByOperation.set(header.operationId, header); + } + + public takeOperationHeader(operationId: string): IDaemonOperationHeaderPayload { + const header: IDaemonOperationHeaderPayload | undefined = + this._headerByOperation.get(operationId); + if (header !== undefined) { + this._headerByOperation.delete(operationId); + this._completedOperations = header.completedOperations; + this._totalOperations = header.totalOperations; + return header; + } + this._completedOperations += OPERATION_COUNT_INCREMENT; + return { + completedOperations: this._completedOperations, + operationId, + totalOperations: this._totalOperations + }; + } +} diff --git a/libraries/rush-terminal-renderer/src/OperationStreamRegistry.ts b/libraries/rush-terminal-renderer/src/OperationStreamRegistry.ts index d761d07a55..8a6cf484ba 100644 --- a/libraries/rush-terminal-renderer/src/OperationStreamRegistry.ts +++ b/libraries/rush-terminal-renderer/src/OperationStreamRegistry.ts @@ -2,11 +2,13 @@ // See LICENSE in the project root for license information. import { NewlineKind } from '@rushstack/node-core-library'; +import type { IDaemonOperationHeaderPayload } from '@rushstack/rush-daemon-protocol'; import { CollatedTerminal, StreamCollator } from '@rushstack/stream-collator'; import type { CollatedWriter } from '@rushstack/stream-collator'; import { TextRewriterTransform } from '@rushstack/terminal'; import type { ITerminalChunk, TerminalWritable } from '@rushstack/terminal'; +import { OperationHeaderTracker } from './OperationHeaderTracker'; import { formatDaemonOperationHeader } from './RendererHeader'; /** Options for {@link OperationStreamRegistry}. @beta */ @@ -29,16 +31,12 @@ export interface IOperationStreamRegistryOptions { export class OperationStreamRegistry { private readonly _collator: StreamCollator; private readonly _collatedTerminal: CollatedTerminal; - private readonly _writers: Map; + private readonly _headers: OperationHeaderTracker = new OperationHeaderTracker(); + private readonly _writers: Map = new Map(); private readonly _quiet: boolean; - private _completedOperations: number; - private _totalOperations: number; public constructor(options: IOperationStreamRegistryOptions) { - this._writers = new Map(); this._quiet = options.quiet; - this._completedOperations = 0; - this._totalOperations = 0; const transform: TextRewriterTransform = new TextRewriterTransform({ destination: options.destination, normalizeNewlines: NewlineKind.OsDefault, @@ -53,7 +51,12 @@ export class OperationStreamRegistry { /** Increments the total-operation count shown in headers. */ public registerOperation(): void { - this._totalOperations += 1; + this._headers.registerOperation(); + } + + /** Records engine-authoritative counters before an operation's stream activates. */ + public setOperationHeader(header: IDaemonOperationHeaderPayload): void { + this._headers.setOperationHeader(header); } /** Writes one raw chunk to the operation's collated stream. */ @@ -78,11 +81,13 @@ export class OperationStreamRegistry { if (writer === undefined) { return; } - this._completedOperations += 1; + const counters: IDaemonOperationHeaderPayload = this._headers.takeOperationHeader( + writer.taskName + ); const header: string = formatDaemonOperationHeader( writer.taskName, - this._completedOperations, - this._totalOperations + counters.completedOperations, + counters.totalOperations ); this._collatedTerminal.writeStdoutLine(`\n${header}`); if (!this._quiet) { diff --git a/libraries/rush-terminal-renderer/src/test/RendererOperationHeader.test.ts b/libraries/rush-terminal-renderer/src/test/RendererOperationHeader.test.ts new file mode 100644 index 0000000000..5157af5874 --- /dev/null +++ b/libraries/rush-terminal-renderer/src/test/RendererOperationHeader.test.ts @@ -0,0 +1,66 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { DaemonVerbosity, IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol'; +import { + RUSHD_OPERATION_HEADER, + RUSHD_OPERATION_STREAM_CLOSED +} from '@rushstack/rush-daemon-protocol'; + +import { DaemonRendererHost } from '../DaemonRendererHost'; + +import { TestTerminal } from './TestTerminal'; + +const OPERATION_ID: string = 'op-a'; +const PROTOCOL_MAJOR: number = 0; +const PROTOCOL_MINOR: number = 1; +const EVENT_SEQUENCE: number = 1; +const VERBOSITIES: ReadonlyArray = ['quiet', 'normal', 'verbose', 'debug']; + +const BASE_ENVELOPE = { + eventId: 'event', + privacy: 'public', + protocolVersion: { major: PROTOCOL_MAJOR, minor: PROTOCOL_MINOR }, + required: false, + sequence: EVENT_SEQUENCE, + sessionId: 'session', + source: { packageName: 'test', packageVersion: '0' }, + timestamp: '2026-08-25T00:00:00.000Z' +} as const; + +function extension(name: string, data: unknown): IDaemonEventEnvelope { + return { + ...BASE_ENVELOPE, + payload: { data, name }, + required: true, + type: 'extension' + }; +} + +it('uses authoritative partial-warm header counters at every verbosity', async () => { + for (const verbosity of VERBOSITIES) { + const terminal: TestTerminal = new TestTerminal(); + const host: DaemonRendererHost = new DaemonRendererHost({ terminal, verbosity }); + await host.initializeAsync(); + for (const operationId of [OPERATION_ID, 'warm-op']) { + host.handleEvent({ + ...BASE_ENVELOPE, + payload: { operationId, silent: false }, + type: 'operationRegistered' + }); + } + host.handleEvent( + extension(RUSHD_OPERATION_HEADER, { + completedOperations: EVENT_SEQUENCE, + operationId: OPERATION_ID, + totalOperations: EVENT_SEQUENCE + }) + ); + host.handleLogChunk(OPERATION_ID, 'stderr', new TextEncoder().encode('failure\n')); + host.handleEvent(extension(RUSHD_OPERATION_STREAM_CLOSED, { operationId: OPERATION_ID })); + + expect(terminal.stdout).toContain('1 of 1'); + expect(terminal.stdout).not.toContain('1 of 2'); + await host.closeAsync(); + } +});