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-global-command-context_2026-08-21-18-44.json b/common/changes/@rushstack/rush-daemon/mojazayeri-global-command-context_2026-08-21-18-44.json new file mode 100644 index 0000000000..b78b587253 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-global-command-context_2026-08-21-18-44.json @@ -0,0 +1,10 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon", + "comment": "Add an opt-in isolated execution context for caller-resolved global commands.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon" +} 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/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..2eeebc19bf 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -6,10 +6,15 @@ /// +import * as childProcess from 'node:child_process'; 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 { ITerminal } from '@rushstack/terminal'; import type { Operation } from '@microsoft/rush-lib'; import { RushConfiguration } from '@microsoft/rush-lib'; import type { RushConfigurationProject } from '@microsoft/rush-lib'; @@ -21,6 +26,16 @@ export type CreateWorkspaceEngineComponentsAsync = (options: ICreateWorkspaceEng // @beta export type CreateWorkspaceSessionComponentsAsync = (options: ICreateWorkspaceSessionComponentsOptions) => Promise; +// @beta +export type GlobalCommandExecutor = (context: IGlobalCommandExecutionContext) => Promise; + +// @beta +export class GlobalCommandRequestRouter { + constructor(workspaceSession: IWorkspaceSession); + executeAsync(request: IResolvedGlobalCommandRequest, executor: GlobalCommandExecutor, client: IGlobalCommandRequestClient): Promise; + resolveRequest(options: IResolveGlobalCommandRequestOptions): IResolvedGlobalCommandRequest; +} + // @beta export interface IClassifyWorkspaceInvalidationsOptions { // (undocumented) @@ -46,6 +61,72 @@ export interface ICreateWorkspaceSessionComponentsOptions { readonly rushConfiguration: RushConfiguration; } +// @beta +export interface IGlobalCommandEnvironment { + // (undocumented) + get(name: string): string | undefined; + // (undocumented) + getNames(): ReadonlyArray; + // (undocumented) + toObject(): NodeJS.ProcessEnv; +} + +// @beta +export interface IGlobalCommandExecutionContext { + // (undocumented) + readonly abortSignal: AbortSignal; + // (undocumented) + readonly cwd: string; + // (undocumented) + readonly environment: IGlobalCommandEnvironment; + // (undocumented) + registerDisposable(disposable: AsyncDisposable): void; + // (undocumented) + spawnChild(command: string, args: ReadonlyArray, options?: IGlobalCommandSpawnOptions): childProcess.ChildProcessWithoutNullStreams; + // (undocumented) + readonly terminal: ITerminal; + // (undocumented) + readonly terminalProperties: IGlobalCommandTerminalProperties; + // (undocumented) + readonly workspaceSession: IWorkspaceSession; +} + +// @beta +export interface IGlobalCommandRequestClient { + readonly abortSignal: AbortSignal; + writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; +} + +// @beta +export interface IGlobalCommandRequestResult { + // (undocumented) + readonly aborted: boolean; + // (undocumented) + readonly requestId: string; +} + +// @beta +export interface IGlobalCommandSpawnOptions { + // (undocumented) + readonly environmentOverlay?: Readonly; + // (undocumented) + readonly forwardOutput?: boolean; + // (undocumented) + readonly shell?: boolean | string; + // (undocumented) + readonly windowsHide?: boolean; +} + +// @beta +export interface IGlobalCommandTerminalProperties { + // (undocumented) + readonly columns: number | undefined; + // (undocumented) + readonly isTTY: boolean; + // (undocumented) + readonly supportsColor: boolean; +} + // @beta export interface IMapWorkspaceInvalidationsOptions { // (undocumented) @@ -58,6 +139,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) @@ -75,6 +165,34 @@ export interface IRequestSchedulerAcquireOptions { waitTimeoutMs?: number; } +// @beta +export interface IResolvedGlobalCommandRequest { + // (undocumented) + readonly commandName: string; + // (undocumented) + readonly cwd: string; + // (undocumented) + readonly environment: IGlobalCommandEnvironment; + // (undocumented) + readonly requestId: string; + // (undocumented) + readonly terminal: IGlobalCommandTerminalProperties; +} + +// @beta +export interface IResolveGlobalCommandRequestOptions { + // (undocumented) + readonly commandName: string; + // (undocumented) + readonly cwd: string; + // (undocumented) + readonly environment: Readonly; + // (undocumented) + readonly requestId: string; + // (undocumented) + readonly terminal: IGlobalCommandTerminalProperties; +} + // @beta export interface IRushDaemonHostOptions { readonly createWorkspaceSessionAsync?: WorkspaceSessionFactory; @@ -217,6 +335,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/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..2fec834737 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -27,3 +27,32 @@ 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). + +`GlobalCommandRequestRouter` is the corresponding opt-in boundary for caller-resolved global command logic. It +canonicalizes and confines the request working directory to the workspace, snapshots its environment, creates a +request-scoped terminal with explicit columns/color/TTY properties, and tracks child processes and async resources +through cancellation or disconnect. Concurrent requests never change `process.cwd()`, `process.env`, or daemon +stdin/stdout/stderr; child commands receive cwd, environment, cancellation, and output routing through the injected +execution context. +Executors must cooperatively observe the context abort signal and settle before cancellation completes, ensuring no +caller-owned logic can outlive its request resources. + +The existing `RushCommandLineParser`, `BaseRushAction`, and some built-in/global action helpers still consult or mutate +process-global state. This layer therefore does not pretend that arbitrary existing actions are daemon-safe: the +integration must supply already resolved command logic that consumes `IGlobalCommandExecutionContext`, including +`spawnChild()` for command-local subprocesses. Adapting the complete action surface remains bounded by the open +[rushstack#5895](https://github.com/microsoft/rushstack/issues/5895) engine/action prerequisite work. Exit-code policy, +interactive stdin/raw-mode/PTY support, scheduling classification, and shared-build merging belong to later layers. 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/GlobalCommandExecutionContext.ts b/libraries/rush-daemon/src/GlobalCommandExecutionContext.ts new file mode 100644 index 0000000000..424e96fc54 --- /dev/null +++ b/libraries/rush-daemon/src/GlobalCommandExecutionContext.ts @@ -0,0 +1,317 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as childProcess from 'node:child_process'; +import { EOL } from 'node:os'; + +import { SubprocessTerminator } from '@rushstack/node-core-library'; +import { Terminal, TerminalProviderSeverity } from '@rushstack/terminal'; +import type { ITerminal, ITerminalProvider } from '@rushstack/terminal'; + +import { + createGlobalCommandEnvironment, + type IGlobalCommandEnvironment, + type IGlobalCommandTerminalProperties, + type IResolvedGlobalCommandRequest +} from './GlobalCommandRequest'; +import type { IGlobalCommandRequestClient } from './GlobalCommandRequestClient'; +import type { IWorkspaceSession } from './WorkspaceSession'; + +const MAX_PENDING_TERMINAL_BYTES: number = 1024 * 1024; + +/** + * Options for a request-scoped child process. + * + * @beta + */ +export interface IGlobalCommandSpawnOptions { + readonly environmentOverlay?: Readonly; + readonly forwardOutput?: boolean; + readonly shell?: boolean | string; + readonly windowsHide?: boolean; +} + +/** + * Explicit state supplied to caller-owned global command logic. + * + * @beta + */ +export interface IGlobalCommandExecutionContext { + readonly abortSignal: AbortSignal; + readonly cwd: string; + readonly environment: IGlobalCommandEnvironment; + readonly terminal: ITerminal; + readonly terminalProperties: IGlobalCommandTerminalProperties; + readonly workspaceSession: IWorkspaceSession; + + registerDisposable(disposable: AsyncDisposable): void; + spawnChild( + command: string, + args: ReadonlyArray, + options?: IGlobalCommandSpawnOptions + ): childProcess.ChildProcessWithoutNullStreams; +} + +class OrderedTerminalWriter { + readonly #client: IGlobalCommandRequestClient; + readonly #onFailure: (error: Error) => void; + #closed: boolean = false; + #failure: Error | undefined; + #pendingByteCount: number = 0; + #tail: Promise = Promise.resolve(); + + public constructor(client: IGlobalCommandRequestClient, onFailure: (error: Error) => void) { + this.#client = client; + this.#onFailure = onFailure; + } + + public write(stream: 'stdout' | 'stderr', chunk: Uint8Array): void { + void this.writeAsync(stream, chunk); + } + + public writeAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise { + if (this.#closed) { + throw new Error('The global command terminal is closed.'); + } + this.#pendingByteCount += chunk.byteLength; + if (this.#pendingByteCount > MAX_PENDING_TERMINAL_BYTES) { + this.#pendingByteCount -= chunk.byteLength; + this.#fail(new Error('The global command terminal output exceeded its pending buffer limit.')); + return this.#tail; + } + this.#tail = this.#tail.then(async () => { + if (this.#failure) { + return; + } + try { + await this.#client.writeTerminalChunkAsync(stream, chunk); + } catch (error) { + this.#fail(normalizeError(error)); + } + }); + this.#tail = this.#tail.finally(() => { + this.#pendingByteCount -= chunk.byteLength; + }); + return this.#tail; + } + + public async closeAsync(): Promise { + this.#closed = true; + await this.#tail; + if (this.#failure) { + throw this.#failure; + } + } + + #fail(error: Error): void { + if (!this.#failure) { + this.#failure = error; + this.#onFailure(error); + } + } +} + +class GlobalCommandTerminalProvider implements ITerminalProvider { + readonly #writer: OrderedTerminalWriter; + readonly #textEncoder: InstanceType = new TextEncoder(); + + public readonly eolCharacter: string = EOL; + public readonly supportsColor: boolean; + + public constructor(writer: OrderedTerminalWriter, supportsColor: boolean) { + this.#writer = writer; + this.supportsColor = supportsColor; + } + + public write(data: string, severity: TerminalProviderSeverity): void { + const stream: 'stdout' | 'stderr' = + severity === TerminalProviderSeverity.error || severity === TerminalProviderSeverity.warning + ? 'stderr' + : 'stdout'; + this.#writer.write(stream, this.#textEncoder.encode(data)); + } +} + +interface ITrackedChild { + readonly completion: Promise; +} + +export class GlobalCommandExecutionContext + implements IGlobalCommandExecutionContext, AsyncDisposable +{ + readonly #abortController: AbortController = new AbortController(); + readonly #client: IGlobalCommandRequestClient; + readonly #disposables: AsyncDisposable[] = []; + readonly #onClientAbort: () => void; + readonly #request: IResolvedGlobalCommandRequest; + readonly #trackedChildren: Set = new Set(); + readonly #childCompletionErrors: unknown[] = []; + readonly #childTerminationErrors: unknown[] = []; + readonly #writer: OrderedTerminalWriter; + #closed: boolean = false; + + public readonly terminal: ITerminal; + public readonly workspaceSession: IWorkspaceSession; + + public constructor( + request: IResolvedGlobalCommandRequest, + client: IGlobalCommandRequestClient, + workspaceSession: IWorkspaceSession + ) { + this.#request = request; + this.#client = client; + this.workspaceSession = workspaceSession; + this.#onClientAbort = () => this.#abortController.abort(client.abortSignal.reason); + this.#writer = new OrderedTerminalWriter(client, (error: Error) => + this.#abortController.abort(error) + ); + this.terminal = new Terminal( + new GlobalCommandTerminalProvider(this.#writer, request.terminal.supportsColor) + ); + if (client.abortSignal.aborted) { + this.#onClientAbort(); + } else { + client.abortSignal.addEventListener('abort', this.#onClientAbort, { once: true }); + } + } + + public get abortSignal(): AbortSignal { + return this.#abortController.signal; + } + + public get cwd(): string { + return this.#request.cwd; + } + + public get environment(): IGlobalCommandEnvironment { + return this.#request.environment; + } + + public get terminalProperties(): IGlobalCommandTerminalProperties { + return this.#request.terminal; + } + + public registerDisposable(disposable: AsyncDisposable): void { + this.#throwIfClosed(); + this.#disposables.push(disposable); + } + + public spawnChild( + command: string, + args: ReadonlyArray, + options: IGlobalCommandSpawnOptions = {} + ): childProcess.ChildProcessWithoutNullStreams { + this.#throwIfClosed(); + if (this.abortSignal.aborted) { + throw this.abortSignal.reason ?? new Error('The global command request was aborted.'); + } + const child: childProcess.ChildProcessWithoutNullStreams = childProcess.spawn(command, [...args], { + cwd: this.cwd, + detached: SubprocessTerminator.RECOMMENDED_OPTIONS.detached, + env: createGlobalCommandEnvironment(this.environment, options.environmentOverlay), + shell: options.shell, + stdio: 'pipe', + windowsHide: options.windowsHide + }); + SubprocessTerminator.killProcessTreeOnExit(child, SubprocessTerminator.RECOMMENDED_OPTIONS); + const completion: Promise = this.#trackChildAsync(child) + .catch((error: unknown) => { + this.#childCompletionErrors.push(error); + }); + const trackedChild: ITrackedChild = { completion }; + this.#trackedChildren.add(trackedChild); + void completion.then(() => this.#trackedChildren.delete(trackedChild)); + if (options.forwardOutput !== false) { + this.#forwardChildOutput(child.stdout, 'stdout'); + this.#forwardChildOutput(child.stderr, 'stderr'); + } + return child; + } + + public async [Symbol.asyncDispose](): Promise { + if (this.#closed) { + return; + } + this.#closed = true; + this.#client.abortSignal.removeEventListener('abort', this.#onClientAbort); + this.#abortController.abort(new Error('The global command execution context was disposed.')); + const cleanupErrors: unknown[] = []; + await Promise.all( + Array.from(this.#trackedChildren, ({ completion }) => + collectCleanupErrorAsync(completion, cleanupErrors) + ) + ); + cleanupErrors.push(...this.#childCompletionErrors); + cleanupErrors.push(...this.#childTerminationErrors); + for (const disposable of this.#disposables.reverse()) { + await collectCleanupErrorAsync( + Promise.resolve().then(() => disposable[Symbol.asyncDispose]()), + cleanupErrors + ); + } + await collectCleanupErrorAsync(this.#writer.closeAsync(), cleanupErrors); + throwCleanupErrors(cleanupErrors); + } + + async #trackChildAsync(child: childProcess.ChildProcessWithoutNullStreams): Promise { + const terminateChild = (): void => { + try { + SubprocessTerminator.killProcessTree(child, SubprocessTerminator.RECOMMENDED_OPTIONS); + } catch (error) { + this.#childTerminationErrors.push(error); + child.kill('SIGKILL'); + } + }; + this.abortSignal.addEventListener('abort', terminateChild, { once: true }); + try { + await new Promise((resolve, reject) => { + child.once('error', reject); + child.once('close', () => resolve()); + }); + } finally { + this.abortSignal.removeEventListener('abort', terminateChild); + } + } + + #forwardChildOutput( + source: NodeJS.ReadableStream & { pause(): unknown; resume(): unknown }, + stream: 'stdout' | 'stderr' + ): void { + source.on('data', (chunk: Buffer) => { + source.pause(); + void this.#writer.writeAsync(stream, chunk).then(() => { + if (!this.abortSignal.aborted) { + source.resume(); + } + }); + }); + source.resume(); + } + + #throwIfClosed(): void { + if (this.#closed) { + throw new Error('The global command execution context is closed.'); + } + } +} + +async function collectCleanupErrorAsync(promise: Promise, cleanupErrors: unknown[]): Promise { + try { + await promise; + } catch (error) { + cleanupErrors.push(error); + } +} + +function throwCleanupErrors(cleanupErrors: unknown[]): void { + if (cleanupErrors.length === 1) { + throw cleanupErrors[0]; + } + if (cleanupErrors.length > 1) { + throw new AggregateError(cleanupErrors, 'Failed to clean up global command request resources.'); + } +} + +function normalizeError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} diff --git a/libraries/rush-daemon/src/GlobalCommandRequest.ts b/libraries/rush-daemon/src/GlobalCommandRequest.ts new file mode 100644 index 0000000000..7969d525d6 --- /dev/null +++ b/libraries/rush-daemon/src/GlobalCommandRequest.ts @@ -0,0 +1,210 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as fs from 'node:fs'; +import * as path from 'node:path'; + +import { EnvironmentMap } from '@rushstack/node-core-library'; + +import type { IWorkspaceSession } from './WorkspaceSession'; + +/** + * Terminal properties captured from the client that submitted a global command. + * + * @beta + */ +export interface IGlobalCommandTerminalProperties { + readonly columns: number | undefined; + readonly isTTY: boolean; + readonly supportsColor: boolean; +} + +/** + * An immutable process-environment snapshot. + * + * @beta + */ +export interface IGlobalCommandEnvironment { + get(name: string): string | undefined; + getNames(): ReadonlyArray; + toObject(): NodeJS.ProcessEnv; +} + +/** + * Untrusted values supplied by a command integration before global-command execution. + * + * @beta + */ +export interface IResolveGlobalCommandRequestOptions { + readonly commandName: string; + readonly cwd: string; + readonly environment: Readonly; + readonly requestId: string; + readonly terminal: IGlobalCommandTerminalProperties; +} + +/** + * A validated request that is safe to execute against one warm workspace session. + * + * @beta + */ +export interface IResolvedGlobalCommandRequest { + readonly commandName: string; + readonly cwd: string; + readonly environment: IGlobalCommandEnvironment; + readonly requestId: string; + readonly terminal: IGlobalCommandTerminalProperties; +} + +const REQUEST_SESSION_BY_REQUEST: WeakMap = new WeakMap(); + +class GlobalCommandEnvironment implements IGlobalCommandEnvironment { + readonly #environmentMap: EnvironmentMap; + readonly #names: ReadonlyArray; + + public constructor(environment: Readonly) { + this.#environmentMap = createEnvironmentMap(environment); + this.#names = Object.freeze( + Array.from(this.#environmentMap.entries(), ({ name }) => name).sort(compareEnvironmentNames) + ); + Object.freeze(this); + } + + public get(name: string): string | undefined { + return this.#environmentMap.get(name); + } + + public getNames(): ReadonlyArray { + return this.#names; + } + + public toObject(): NodeJS.ProcessEnv { + return this.#environmentMap.toObject(); + } +} + +export function resolveGlobalCommandRequest( + options: IResolveGlobalCommandRequestOptions, + workspaceSession: IWorkspaceSession +): IResolvedGlobalCommandRequest { + validateNonemptyName(options.requestId, 'request id'); + validateNonemptyName(options.commandName, 'command name'); + const repoRoot: string = getCanonicalDirectory(workspaceSession.metadata.repoRoot, 'workspace root'); + const cwd: string = getCanonicalDirectory(options.cwd, 'working directory'); + validatePathWithinWorkspace(cwd, repoRoot); + const request: IResolvedGlobalCommandRequest = Object.freeze({ + commandName: options.commandName, + cwd, + environment: new GlobalCommandEnvironment(options.environment), + requestId: options.requestId, + terminal: resolveTerminalProperties(options.terminal) + }); + REQUEST_SESSION_BY_REQUEST.set(request, workspaceSession); + return request; +} + +export function validateResolvedGlobalCommandRequest( + request: IResolvedGlobalCommandRequest, + workspaceSession: IWorkspaceSession +): void { + if (REQUEST_SESSION_BY_REQUEST.get(request) !== workspaceSession) { + throw new Error('The global command request was not resolved for this workspace session.'); + } +} + +export function createGlobalCommandEnvironment( + baseEnvironment: IGlobalCommandEnvironment, + overlay: Readonly | undefined +): NodeJS.ProcessEnv { + const environmentMap: EnvironmentMap = new EnvironmentMap(baseEnvironment.toObject()); + if (overlay) { + for (const [name, value] of Object.entries(overlay)) { + validateEnvironmentName(name); + if (value === undefined) { + environmentMap.unset(name); + } else { + validateEnvironmentValue(name, value); + environmentMap.set(name, value); + } + } + } + return environmentMap.toObject(); +} + +function createEnvironmentMap(environment: Readonly): EnvironmentMap { + const environmentMap: EnvironmentMap = new EnvironmentMap(); + for (const [name, value] of Object.entries(environment)) { + validateEnvironmentName(name); + if (value !== undefined) { + validateEnvironmentValue(name, value); + environmentMap.set(name, value); + } + } + return environmentMap; +} + +function validateEnvironmentName(name: string): void { + if (name.length === 0 || name.includes('=') || name.includes('\0')) { + throw new Error(`Invalid global command environment variable name: "${name}".`); + } +} + +function validateEnvironmentValue(name: string, value: unknown): asserts value is string { + if (typeof value !== 'string') { + throw new Error(`The global command environment variable "${name}" must have a string value.`); + } + if (value.includes('\0')) { + throw new Error(`The global command environment variable "${name}" contains a null character.`); + } +} + +function resolveTerminalProperties( + terminal: IGlobalCommandTerminalProperties +): IGlobalCommandTerminalProperties { + if ( + terminal.columns !== undefined && + (!Number.isSafeInteger(terminal.columns) || terminal.columns <= 0) + ) { + throw new Error('Global command terminal columns must be a positive safe integer.'); + } + if (typeof terminal.isTTY !== 'boolean' || typeof terminal.supportsColor !== 'boolean') { + throw new Error('Global command terminal TTY and color properties must be boolean values.'); + } + return Object.freeze({ + columns: terminal.columns, + isTTY: terminal.isTTY, + supportsColor: terminal.supportsColor + }); +} + +function getCanonicalDirectory(folderPath: string, kind: string): string { + let canonicalPath: string; + try { + canonicalPath = fs.realpathSync.native(path.resolve(folderPath)); + } catch (error) { + throw new Error(`The global command ${kind} does not resolve to an existing directory: ${folderPath}`, { + cause: error + }); + } + if (!fs.statSync(canonicalPath).isDirectory()) { + throw new Error(`The global command ${kind} is not a directory: ${folderPath}`); + } + return canonicalPath; +} + +function validatePathWithinWorkspace(cwd: string, repoRoot: string): void { + const relativePath: string = path.relative(repoRoot, cwd); + if (relativePath === '..' || relativePath.startsWith(`..${path.sep}`) || path.isAbsolute(relativePath)) { + throw new Error(`The global command working directory is outside the daemon workspace: ${cwd}`); + } +} + +function validateNonemptyName(value: string, kind: string): void { + if (value.length === 0 || value.trim() !== value) { + throw new Error(`Invalid global command ${kind}: "${value}".`); + } +} + +function compareEnvironmentNames(left: string, right: string): number { + return left.localeCompare(right); +} diff --git a/libraries/rush-daemon/src/GlobalCommandRequestClient.ts b/libraries/rush-daemon/src/GlobalCommandRequestClient.ts new file mode 100644 index 0000000000..a7346509d9 --- /dev/null +++ b/libraries/rush-daemon/src/GlobalCommandRequestClient.ts @@ -0,0 +1,19 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** + * A client-scoped destination for one global command request. + * + * @remarks + * The client must abort `abortSignal` when its request is cancelled or its connection closes. Terminal writes are + * serialized in command order, and each promise provides the destination's backpressure boundary. + * + * @beta + */ +export interface IGlobalCommandRequestClient { + /** Aborted by the transport when the request is cancelled or disconnected. */ + readonly abortSignal: AbortSignal; + + /** Writes one request-scoped terminal chunk through the client's backpressured destination. */ + writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; +} diff --git a/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts b/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts new file mode 100644 index 0000000000..9ef996bb27 --- /dev/null +++ b/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts @@ -0,0 +1,135 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IGlobalCommandExecutionContext } from './GlobalCommandExecutionContext'; +import { GlobalCommandExecutionContext } from './GlobalCommandExecutionContext'; +import { + type IResolvedGlobalCommandRequest, + type IResolveGlobalCommandRequestOptions, + resolveGlobalCommandRequest, + validateResolvedGlobalCommandRequest +} from './GlobalCommandRequest'; +import type { IGlobalCommandRequestClient } from './GlobalCommandRequestClient'; +import type { IWorkspaceSession } from './WorkspaceSession'; + +/** + * Executes caller-resolved global command logic. + * + * @remarks + * The executor must observe `context.abortSignal` and settle before cancellation can complete. This preserves the + * invariant that no caller-owned command logic remains active after the request context is cleaned up. + * + * @beta + */ +export type GlobalCommandExecutor = (context: IGlobalCommandExecutionContext) => Promise; + +/** + * The completion state for one global command request. + * + * @beta + */ +export interface IGlobalCommandRequestResult { + readonly aborted: boolean; + readonly requestId: string; +} + +/** + * Routes caller-owned global command logic through isolated per-request process and terminal context. + * + * @remarks + * Command parsing and Rush action construction remain integration-owned. This router intentionally does not invoke + * existing Rush actions that still depend on process-global cwd, environment, or console state. + * + * @beta + */ +export class GlobalCommandRequestRouter { + readonly #workspaceSession: IWorkspaceSession; + + public constructor(workspaceSession: IWorkspaceSession) { + this.#workspaceSession = workspaceSession; + } + + /** Validates and snapshots an untrusted global request for this workspace. */ + public resolveRequest(options: IResolveGlobalCommandRequestOptions): IResolvedGlobalCommandRequest { + return resolveGlobalCommandRequest(options, this.#workspaceSession); + } + + /** Executes an already resolved global request without mutating daemon process globals. */ + public async executeAsync( + request: IResolvedGlobalCommandRequest, + executor: GlobalCommandExecutor, + client: IGlobalCommandRequestClient + ): Promise { + validateResolvedGlobalCommandRequest(request, this.#workspaceSession); + const context: GlobalCommandExecutionContext = new GlobalCommandExecutionContext( + request, + client, + this.#workspaceSession + ); + let executionError: unknown; + let aborted: boolean = context.abortSignal.aborted; + try { + if (!aborted) { + const executorPromise: Promise = Promise.resolve().then(() => executor(context)); + const outcome: 'aborted' | 'completed' = await waitForExecutionAsync( + executorPromise, + context.abortSignal + ); + aborted = outcome === 'aborted'; + } + } catch (error) { + executionError = error; + } + + let cleanupError: unknown; + try { + await context[Symbol.asyncDispose](); + } catch (error) { + cleanupError = error; + } + throwExecutionAndCleanupErrors(executionError, cleanupError); + return { aborted, requestId: request.requestId }; + } +} + +async function waitForExecutionAsync( + executorPromise: Promise, + abortSignal: AbortSignal +): Promise<'aborted' | 'completed'> { + let removeAbortListener: (() => void) | undefined; + const abortPromise: Promise<'aborted'> = new Promise((resolve) => { + const onAbort = (): void => resolve('aborted'); + removeAbortListener = () => abortSignal.removeEventListener('abort', onAbort); + if (abortSignal.aborted) { + resolve('aborted'); + } else { + abortSignal.addEventListener('abort', onAbort, { once: true }); + } + }); + const completedPromise: Promise<'completed'> = executorPromise.then(() => 'completed'); + try { + const outcome: 'aborted' | 'completed' = await Promise.race([completedPromise, abortPromise]); + if (outcome === 'aborted') { + await executorPromise.catch(() => undefined); + } + return outcome; + } finally { + removeAbortListener?.(); + void executorPromise.catch(() => undefined); + } +} + +function throwExecutionAndCleanupErrors(executionError: unknown, cleanupError: unknown): void { + if (executionError !== undefined && cleanupError !== undefined) { + throw new AggregateError( + [executionError, cleanupError], + 'The global command failed and could not clean up its request context.' + ); + } + if (executionError !== undefined) { + throw executionError; + } + if (cleanupError !== undefined) { + throw cleanupError; + } +} 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..7d52e37841 --- /dev/null +++ b/libraries/rush-daemon/src/PhasedRequestEventSink.ts @@ -0,0 +1,204 @@ +// 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 errorMessage: string | undefined; + 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, { + errorMessage: result.error?.message, + 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 + }); + } + } + + 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 || !this.#activeOperationIds.has(operationId)) { + return; + } + this.#emitEvent( + 'activityChanged', + { stream: options?.stderr === true ? 'stderr' : 'stdout', text }, + { required: true, scope: { 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..183f43cf9f --- /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.errorMessage + : 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..a95b8b766f 100644 --- a/libraries/rush-daemon/src/index.ts +++ b/libraries/rush-daemon/src/index.ts @@ -11,6 +11,22 @@ export { RequestSchedulerError, RequestSchedulerErrorCode } from './RequestScheduler'; +export { + type IGlobalCommandExecutionContext, + type IGlobalCommandSpawnOptions +} from './GlobalCommandExecutionContext'; +export { + type IGlobalCommandEnvironment, + type IGlobalCommandTerminalProperties, + type IResolvedGlobalCommandRequest, + type IResolveGlobalCommandRequestOptions +} from './GlobalCommandRequest'; +export { type IGlobalCommandRequestClient } from './GlobalCommandRequestClient'; +export { + type GlobalCommandExecutor, + GlobalCommandRequestRouter, + type IGlobalCommandRequestResult +} from './GlobalCommandRequestRouter'; export { RushDaemonHost, type IRushDaemonHostOptions } from './RushDaemonHost'; export { serveRushDaemonAsync, type IRushDaemonServeOptions } from './serveRushDaemon'; export { @@ -42,3 +58,5 @@ export { WorkspaceInvalidationTracker, type IWorkspaceInvalidationSnapshot } from './WorkspaceInvalidationTracker'; +export { type IPhasedRequestClient } from './PhasedRequestClient'; +export { PhasedRequestRouter } from './PhasedRequestRouter'; diff --git a/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts b/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts new file mode 100644 index 0000000000..f5bd82b4d6 --- /dev/null +++ b/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts @@ -0,0 +1,447 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as fs from 'node:fs'; +import * as path from 'node:path'; + +import { SubprocessTerminator } from '@rushstack/node-core-library'; + +import type { IGlobalCommandExecutionContext } from '../GlobalCommandExecutionContext'; +import type { + IResolvedGlobalCommandRequest, + IResolveGlobalCommandRequestOptions +} from '../GlobalCommandRequest'; +import type { IGlobalCommandRequestClient } from '../GlobalCommandRequestClient'; +import { + GlobalCommandRequestRouter, + type IGlobalCommandRequestResult +} from '../GlobalCommandRequestRouter'; +import { TestWorkspaceSession, TEST_REPO_ROOT } from './TestWorkspaceSession'; + +const TEXT_DECODER: InstanceType = new TextDecoder(); +const FIRST_CWD: string = path.join(TEST_REPO_ROOT, 'libraries', 'rush-daemon'); +const SECOND_CWD: string = path.join(TEST_REPO_ROOT, 'libraries', 'terminal'); + +interface IClientChunk { + readonly stream: 'stdout' | 'stderr'; + readonly text: string; +} + +class TestGlobalCommandClient implements IGlobalCommandRequestClient { + public readonly abortController: AbortController = new AbortController(); + public readonly chunks: IClientChunk[] = []; + public onWriteAsync: ((chunk: IClientChunk) => Promise) | undefined; + + public get abortSignal(): AbortSignal { + return this.abortController.signal; + } + + public async writeTerminalChunkAsync( + stream: 'stdout' | 'stderr', + chunk: Uint8Array + ): Promise { + const clientChunk: IClientChunk = { stream, text: TEXT_DECODER.decode(chunk) }; + this.chunks.push(clientChunk); + await this.onWriteAsync?.(clientChunk); + } +} + +function createRequestOptions( + requestId: string, + cwd: string, + environment: Readonly, + columns: number +): IResolveGlobalCommandRequestOptions { + return { + commandName: 'global-test', + cwd, + environment, + requestId, + terminal: { columns, isTTY: true, supportsColor: columns > 100 } + }; +} + +function getCanonicalPath(folderPath: string): string { + return fs.realpathSync.native(folderPath); +} + +function waitForAbortAsync(signal: AbortSignal): Promise { + return new Promise((resolve) => { + if (signal.aborted) { + resolve(); + } else { + signal.addEventListener('abort', () => resolve(), { once: true }); + } + }); +} + +function createRecordingDisposable(name: string, disposalOrder: string[]): AsyncDisposable { + return { + [Symbol.asyncDispose]: (): Promise => { + disposalOrder.push(name); + return Promise.resolve(); + } + }; +} + +describe(GlobalCommandRequestRouter.name, () => { + it('isolates concurrent cwd, environment, and terminal state without changing daemon globals', async () => { + const processCwd: string = process.cwd(); + const processEnvironmentValue: string | undefined = process.env.RUSHD_CONTEXT_TEST; + const session: TestWorkspaceSession = new TestWorkspaceSession(TEST_REPO_ROOT); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + const observations: string[] = []; + let releaseExecutors: (() => void) | undefined; + let startedCount: number = 0; + const executorsStarted: Promise = new Promise((resolve) => { + releaseExecutors = resolve; + }); + const runAsync = async ( + request: IResolvedGlobalCommandRequest, + client: TestGlobalCommandClient + ): Promise => + router.executeAsync( + request, + async (context: IGlobalCommandExecutionContext): Promise => { + observations.push( + [ + context.cwd, + context.environment.get('RUSHD_CONTEXT_TEST'), + context.terminalProperties.columns, + context.terminalProperties.supportsColor + ].join('|') + ); + context.terminal.writeLine(request.requestId); + if (++startedCount === 2) { + releaseExecutors?.(); + } + await executorsStarted; + expect(process.cwd()).toBe(processCwd); + expect(process.env.RUSHD_CONTEXT_TEST).toBe(processEnvironmentValue); + }, + client + ); + const firstClient: TestGlobalCommandClient = new TestGlobalCommandClient(); + const secondClient: TestGlobalCommandClient = new TestGlobalCommandClient(); + const firstRequest: IResolvedGlobalCommandRequest = router.resolveRequest( + createRequestOptions('first', FIRST_CWD, { RUSHD_CONTEXT_TEST: 'first' }, 80) + ); + const secondRequest: IResolvedGlobalCommandRequest = router.resolveRequest( + createRequestOptions('second', SECOND_CWD, { RUSHD_CONTEXT_TEST: 'second' }, 160) + ); + + const results: IGlobalCommandRequestResult[] = await Promise.all([ + runAsync(firstRequest, firstClient), + runAsync(secondRequest, secondClient) + ]); + + expect(results).toEqual([ + { aborted: false, requestId: 'first' }, + { aborted: false, requestId: 'second' } + ]); + expect(new Set(observations)).toEqual( + new Set([ + `${getCanonicalPath(FIRST_CWD)}|first|80|false`, + `${getCanonicalPath(SECOND_CWD)}|second|160|true` + ]) + ); + expect(firstClient.chunks.map(({ text }) => text).join('')).toContain('first'); + expect(secondClient.chunks.map(({ text }) => text).join('')).toContain('second'); + expect(process.cwd()).toBe(processCwd); + expect(process.env.RUSHD_CONTEXT_TEST).toBe(processEnvironmentValue); + }); + + it('snapshots request environment and propagates isolated context to child processes', async () => { + const mutableEnvironment: NodeJS.ProcessEnv = { + CHILD_CONTEXT: 'request', + REMOVED_CONTEXT: 'remove-me' + }; + const session: TestWorkspaceSession = new TestWorkspaceSession(TEST_REPO_ROOT); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + const request: IResolvedGlobalCommandRequest = router.resolveRequest( + createRequestOptions('spawn', FIRST_CWD, mutableEnvironment, 80) + ); + mutableEnvironment.CHILD_CONTEXT = 'mutated-after-resolution'; + const copiedEnvironment: NodeJS.ProcessEnv = request.environment.toObject(); + copiedEnvironment.CHILD_CONTEXT = 'mutated-copy'; + const client: TestGlobalCommandClient = new TestGlobalCommandClient(); + + await router.executeAsync( + request, + async (context: IGlobalCommandExecutionContext): Promise => { + const child = context.spawnChild( + process.execPath, + [ + '-e', + 'process.stdout.write(JSON.stringify({cwd:process.cwd(),value:process.env.CHILD_CONTEXT,removed:process.env.REMOVED_CONTEXT}))' + ], + { + environmentOverlay: { + CHILD_CONTEXT: `${context.environment.get('CHILD_CONTEXT')}-child`, + REMOVED_CONTEXT: undefined + } + } + ); + await new Promise((resolve, reject) => { + child.once('error', reject); + child.once('close', () => resolve()); + }); + }, + client + ); + + const childOutput: { cwd: string; removed?: string; value: string } = JSON.parse( + client.chunks + .filter(({ stream }) => stream === 'stdout') + .map(({ text }) => text) + .join('') + ); + expect(childOutput).toEqual({ + cwd: getCanonicalPath(FIRST_CWD), + value: 'request-child' + }); + expect(request.environment.get('CHILD_CONTEXT')).toBe('request'); + }); + + it('reports child spawn failures during request cleanup', async () => { + const session: TestWorkspaceSession = new TestWorkspaceSession(TEST_REPO_ROOT); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + + await expect( + router.executeAsync( + router.resolveRequest(createRequestOptions('spawn-failure', FIRST_CWD, {}, 80)), + async (context: IGlobalCommandExecutionContext): Promise => { + context.spawnChild(path.join(FIRST_CWD, 'missing-global-command'), [], { + forwardOutput: false + }); + await new Promise((resolve) => setImmediate(resolve)); + }, + new TestGlobalCommandClient() + ) + ).rejects.toThrow(/ENOENT|spawn/); + }); + + it('rejects non-string values in untrusted environment snapshots and overlays', async () => { + const session: TestWorkspaceSession = new TestWorkspaceSession(TEST_REPO_ROOT); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + const invalidEnvironment: NodeJS.ProcessEnv = JSON.parse('{"INVALID_VALUE":123}') as NodeJS.ProcessEnv; + + expect(() => + router.resolveRequest(createRequestOptions('invalid-environment', FIRST_CWD, invalidEnvironment, 80)) + ).toThrow('environment variable "INVALID_VALUE" must have a string value'); + + const invalidOverlay: NodeJS.ProcessEnv = JSON.parse('{"INVALID_OVERLAY":false}') as NodeJS.ProcessEnv; + await expect( + router.executeAsync( + router.resolveRequest(createRequestOptions('invalid-overlay', FIRST_CWD, {}, 80)), + async (context: IGlobalCommandExecutionContext): Promise => { + context.spawnChild(process.execPath, [], { environmentOverlay: invalidOverlay }); + }, + new TestGlobalCommandClient() + ) + ).rejects.toThrow('environment variable "INVALID_OVERLAY" must have a string value'); + }); + + it('cleans registered resources after success and failure without disposing the warm session', async () => { + let sessionDisposeCount: number = 0; + let requestDisposeCount: number = 0; + const session: TestWorkspaceSession = new TestWorkspaceSession( + TEST_REPO_ROOT, + () => sessionDisposeCount++ + ); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + const registerDisposable = (context: IGlobalCommandExecutionContext): void => { + context.registerDisposable({ + [Symbol.asyncDispose]: (): Promise => { + requestDisposeCount++; + return Promise.resolve(); + } + }); + }; + + await router.executeAsync( + router.resolveRequest(createRequestOptions('success', FIRST_CWD, {}, 80)), + async (context: IGlobalCommandExecutionContext): Promise => registerDisposable(context), + new TestGlobalCommandClient() + ); + await expect( + router.executeAsync( + router.resolveRequest(createRequestOptions('failure', SECOND_CWD, {}, 80)), + async (context: IGlobalCommandExecutionContext): Promise => { + registerDisposable(context); + throw new Error('global command failed'); + }, + new TestGlobalCommandClient() + ) + ).rejects.toThrow('global command failed'); + + expect(requestDisposeCount).toBe(2); + expect(sessionDisposeCount).toBe(0); + }); + + it('aborts child processes and cleans request resources on cancellation', async () => { + const processCwd: string = process.cwd(); + const processEnvironmentValue: string | undefined = process.env.RUSHD_CONTEXT_TEST; + const session: TestWorkspaceSession = new TestWorkspaceSession(TEST_REPO_ROOT); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + const client: TestGlobalCommandClient = new TestGlobalCommandClient(); + const killProcessTreeSpy: jest.SpyInstance = jest.spyOn( + SubprocessTerminator, + 'killProcessTree' + ); + const killProcessTreeOnExitSpy: jest.SpyInstance = jest.spyOn( + SubprocessTerminator, + 'killProcessTreeOnExit' + ); + let resourceDisposed: boolean = false; + let markChildStarted: (() => void) | undefined; + const childStarted: Promise = new Promise((resolve) => { + markChildStarted = resolve; + }); + const resultPromise: Promise = router.executeAsync( + router.resolveRequest( + createRequestOptions('cancelled', FIRST_CWD, { RUSHD_CONTEXT_TEST: 'child' }, 80) + ), + async (context: IGlobalCommandExecutionContext): Promise => { + context.registerDisposable({ + [Symbol.asyncDispose]: (): Promise => { + resourceDisposed = true; + return Promise.resolve(); + } + }); + const child = context.spawnChild(process.execPath, ['-e', 'setInterval(() => {}, 1000)']); + child.once('spawn', () => markChildStarted?.()); + await new Promise((resolve) => child.once('close', () => resolve())); + }, + client + ); + try { + await childStarted; + client.abortController.abort(new Error('client cancelled')); + + await expect(resultPromise).resolves.toEqual({ aborted: true, requestId: 'cancelled' }); + expect(resourceDisposed).toBe(true); + expect(killProcessTreeOnExitSpy).toHaveBeenCalledTimes(1); + expect(killProcessTreeSpy).toHaveBeenCalledTimes(1); + expect(process.cwd()).toBe(processCwd); + expect(process.env.RUSHD_CONTEXT_TEST).toBe(processEnvironmentValue); + } finally { + killProcessTreeOnExitSpy.mockRestore(); + killProcessTreeSpy.mockRestore(); + } + }); + + it('waits for cooperative executor settlement before completing cancellation', async () => { + const session: TestWorkspaceSession = new TestWorkspaceSession(TEST_REPO_ROOT); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + const client: TestGlobalCommandClient = new TestGlobalCommandClient(); + let releaseExecutor: (() => void) | undefined; + let executorSettled: boolean = false; + const executorRelease: Promise = new Promise((resolve) => { + releaseExecutor = resolve; + }); + const resultPromise: Promise = router.executeAsync( + router.resolveRequest(createRequestOptions('cooperative-cancel', FIRST_CWD, {}, 80)), + async (context: IGlobalCommandExecutionContext): Promise => { + await waitForAbortAsync(context.abortSignal); + await executorRelease; + executorSettled = true; + }, + client + ); + let requestSettled: boolean = false; + void resultPromise.then(() => { + requestSettled = true; + }); + + client.abortController.abort(); + await new Promise((resolve) => setImmediate(resolve)); + expect(requestSettled).toBe(false); + releaseExecutor?.(); + + await expect(resultPromise).resolves.toEqual({ + aborted: true, + requestId: 'cooperative-cancel' + }); + expect(executorSettled).toBe(true); + }); + + it('continues request cleanup after a disposer throws synchronously', async () => { + const session: TestWorkspaceSession = new TestWorkspaceSession(TEST_REPO_ROOT); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + const disposalOrder: string[] = []; + + await expect( + router.executeAsync( + router.resolveRequest(createRequestOptions('cleanup-errors', FIRST_CWD, {}, 80)), + async (context: IGlobalCommandExecutionContext): Promise => { + context.registerDisposable(createRecordingDisposable('first', disposalOrder)); + context.registerDisposable({ + [Symbol.asyncDispose]: (): Promise => { + disposalOrder.push('throwing'); + throw new Error('synchronous cleanup failure'); + } + }); + context.registerDisposable(createRecordingDisposable('last', disposalOrder)); + }, + new TestGlobalCommandClient() + ) + ).rejects.toThrow('synchronous cleanup failure'); + expect(disposalOrder).toEqual(['last', 'throwing', 'first']); + }); + + it('surfaces disconnect write failures after deterministic cleanup', async () => { + const session: TestWorkspaceSession = new TestWorkspaceSession(TEST_REPO_ROOT); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); + const client: TestGlobalCommandClient = new TestGlobalCommandClient(); + let resourceDisposed: boolean = false; + client.onWriteAsync = (): Promise => Promise.reject(new Error('client disconnected')); + + await expect( + router.executeAsync( + router.resolveRequest(createRequestOptions('disconnect', FIRST_CWD, {}, 80)), + async (context: IGlobalCommandExecutionContext): Promise => { + context.registerDisposable({ + [Symbol.asyncDispose]: (): Promise => { + resourceDisposed = true; + return Promise.resolve(); + } + }); + context.terminal.writeLine('disconnect'); + await waitForAbortAsync(context.abortSignal); + }, + client + ) + ).rejects.toThrow('client disconnected'); + expect(resourceDisposed).toBe(true); + }); + + it('rejects invalid or cross-workspace resolved requests before execution', async () => { + const firstRouter: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const secondRouter: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + expect(() => + firstRouter.resolveRequest(createRequestOptions('outside', path.dirname(TEST_REPO_ROOT), {}, 80)) + ).toThrow('outside the daemon workspace'); + expect(() => + firstRouter.resolveRequest(createRequestOptions('columns', FIRST_CWD, {}, 0)) + ).toThrow('positive safe integer'); + const request: IResolvedGlobalCommandRequest = firstRouter.resolveRequest( + createRequestOptions('first-workspace', FIRST_CWD, {}, 80) + ); + const executor: jest.Mock, [IGlobalCommandExecutionContext]> = jest.fn( + (context: IGlobalCommandExecutionContext) => { + void context; + return Promise.resolve(); + } + ); + + await expect( + secondRouter.executeAsync(request, executor, new TestGlobalCommandClient()) + ).rejects.toThrow('not resolved for this workspace session'); + expect(executor).not.toHaveBeenCalled(); + }); +}); 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..62d88d425f --- /dev/null +++ b/libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts @@ -0,0 +1,531 @@ +// 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_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'); + } + }); + 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]); + }); +}); 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) + }; +}