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/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index cec673082f..2eeebc19bf 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -6,6 +6,7 @@ /// +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'; @@ -13,6 +14,7 @@ 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'; @@ -24,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) @@ -49,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) @@ -87,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; diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index adc605597d..2fec834737 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -40,3 +40,19 @@ This layer deliberately does not add control-frame admission or reconstruct `Pha 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/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/index.ts b/libraries/rush-daemon/src/index.ts index a49fb7bad6..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 { 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(); + }); +});