diff --git a/common/changes/@rushstack/rush-daemon/mojazayeri-warm-graph-components_2026-08-19-22-30.json b/common/changes/@rushstack/rush-daemon/mojazayeri-warm-graph-components_2026-08-19-22-30.json new file mode 100644 index 0000000000..ca05af3ff0 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-warm-graph-components_2026-08-19-22-30.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon", + "comment": "Add an opt-in all-project engine component factory with explicit phase/plugin shape, retained invalidation reconciliation, and a deterministic engine shutdown contract.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-daemon/mojazayeri-warm-workspace-session_2026-08-18-22-30.json b/common/changes/@rushstack/rush-daemon/mojazayeri-warm-workspace-session_2026-08-18-22-30.json new file mode 100644 index 0000000000..02d7a0df55 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-warm-workspace-session_2026-08-18-22-30.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon", + "comment": "Add a reusable warm workspace session with stable Rush configuration metadata, retained headless invalidations, and deterministic host lifecycle integration.", + "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 a0167a937a..74a7885c75 100644 --- a/common/config/subspaces/default/pnpm-lock.yaml +++ b/common/config/subspaces/default/pnpm-lock.yaml @@ -4079,6 +4079,9 @@ importers: ../../../libraries/rush-daemon: dependencies: + '@microsoft/rush-lib': + specifier: workspace:* + version: link:../rush-lib '@rushstack/node-core-library': specifier: workspace:* version: link:../node-core-library diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index ef178340ec..b0b4ae9594 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -6,7 +6,49 @@ /// +import type { GetInputsSnapshotAsyncFn } from '@microsoft/rush-lib'; import type { IDaemonPaths } from '@rushstack/rush-daemon-transport'; +import type { IInputsSnapshot } from '@microsoft/rush-lib'; +import type { IOperationGraph } from '@microsoft/rush-lib'; +import type { Operation } from '@microsoft/rush-lib'; +import { RushConfiguration } from '@microsoft/rush-lib'; +import type { RushConfigurationProject } from '@microsoft/rush-lib'; +import type { RushSession } from '@microsoft/rush-lib'; + +// @beta +export type CreateWorkspaceEngineComponentsAsync = (options: ICreateWorkspaceEngineComponentsOptions) => Promise; + +// @beta +export type CreateWorkspaceSessionComponentsAsync = (options: ICreateWorkspaceSessionComponentsOptions) => Promise; + +// @beta +export interface ICreateWorkspaceEngineComponentsOptions extends IWorkspaceEngineShape { + readonly projectSelection: ReadonlySet; + // (undocumented) + readonly rushConfiguration: RushConfiguration; +} + +// @beta +export interface ICreateWorkspaceSessionComponentsOptions { + // (undocumented) + readonly invalidations: WorkspaceInvalidationTracker; + // (undocumented) + readonly onError?: (error: Error) => void; + // (undocumented) + readonly rushConfiguration: RushConfiguration; +} + +// @beta +export interface IMapWorkspaceInvalidationsOptions { + // (undocumented) + readonly changedPaths: ReadonlyArray; + // (undocumented) + readonly currentInputsSnapshot: IInputsSnapshot; + // (undocumented) + readonly nextInputsSnapshot: IInputsSnapshot; + // (undocumented) + readonly operationGraph: IOperationGraph; +} // @public export interface IRequestLease { @@ -27,6 +69,7 @@ export interface IRequestSchedulerAcquireOptions { // @beta export interface IRushDaemonHostOptions { + readonly createWorkspaceSessionAsync?: WorkspaceSessionFactory; readonly daemonVersion: string; readonly onError?: (error: Error) => void; readonly repoRoot: string; @@ -40,6 +83,127 @@ export interface IRushDaemonServeOptions extends IRushDaemonHostOptions { readonly shutdownSignal?: AbortSignal; } +// @beta +export interface IWorkspaceEngineComponentFactoryOptions { + // (undocumented) + readonly createEngineComponentsAsync: CreateWorkspaceEngineComponentsAsync; + // (undocumented) + readonly mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; + // (undocumented) + readonly shape: IWorkspaceEngineShape; +} + +// @beta +export interface IWorkspaceEngineComponents extends AsyncDisposable { + [Symbol.asyncDispose](): Promise; + // (undocumented) + readonly getInputsSnapshotAsync: GetInputsSnapshotAsyncFn; + // (undocumented) + readonly inputsSnapshot: IInputsSnapshot; + // (undocumented) + readonly operationGraph: IOperationGraph; + // (undocumented) + readonly rushSession: RushSession; +} + +// @beta +export interface IWorkspaceEngineShape { + // (undocumented) + readonly phaseNames: ReadonlyArray; + // (undocumented) + readonly pluginNames: ReadonlyArray; +} + +// @beta +export interface IWorkspaceInvalidationReconciliation { + // (undocumented) + readonly inputsSnapshot: IInputsSnapshot; + // (undocumented) + readonly invalidatedOperationCount: number; + // (undocumented) + readonly isFullInvalidation: boolean; + // (undocumented) + readonly sequence: number; +} + +// @beta +export interface IWorkspaceInvalidationSnapshot { + readonly changedPaths: ReadonlyArray; + readonly hasUnknownChanges: boolean; + readonly isWatcherHealthy: boolean; + readonly sequence: number; +} + +// @beta +export interface IWorkspaceInvalidationWatcher extends AsyncDisposable { + // (undocumented) + startAsync(onInvalidation: (changedPath?: string) => void): Promise; +} + +// @beta +export interface IWorkspaceSession extends AsyncDisposable { + // (undocumented) + readonly engineShape: IWorkspaceEngineShape | undefined; + // (undocumented) + readonly inputsSnapshot: IInputsSnapshot | undefined; + // (undocumented) + readonly invalidations: WorkspaceInvalidationTracker; + // (undocumented) + readonly metadata: IWorkspaceSessionMetadata; + // (undocumented) + readonly operationGraph: IOperationGraph | undefined; + // (undocumented) + reconcileInvalidationsAsync(): Promise; + // (undocumented) + readonly rushConfiguration: RushConfiguration; + // (undocumented) + readonly rushSession: RushSession | undefined; +} + +// @beta +export interface IWorkspaceSessionComponents extends AsyncDisposable { + // (undocumented) + readonly engineShape?: IWorkspaceEngineShape; + // (undocumented) + readonly inputsSnapshot?: IInputsSnapshot; + // (undocumented) + readonly operationGraph?: IOperationGraph; + readonly projectWatcher?: IWorkspaceInvalidationWatcher; + // (undocumented) + readonly reconcileInvalidationsAsync?: () => Promise; + // (undocumented) + readonly rushSession?: RushSession; +} + +// @beta +export interface IWorkspaceSessionMetadata { + // (undocumented) + readonly projectCount: number; + // (undocumented) + readonly projectNames: ReadonlyArray; + // (undocumented) + readonly repoRoot: string; + // (undocumented) + readonly rushJsonFile: string; + // (undocumented) + readonly rushVersion: string; +} + +// @beta +export interface IWorkspaceSessionOptions { + // (undocumented) + readonly createComponentsAsync?: CreateWorkspaceSessionComponentsAsync; + // (undocumented) + readonly onError?: (error: Error) => void; + // (undocumented) + readonly repoRoot: string; + // (undocumented) + readonly rushVersion: string; +} + +// @beta +export type MapWorkspaceInvalidationsToOperationsAsync = (options: IMapWorkspaceInvalidationsOptions) => Promise>; + // @public export enum RequestExclusivityClass { // (undocumented) @@ -77,6 +241,7 @@ export enum RequestSchedulerErrorCode { // @beta export class RushDaemonHost { closeAsync(): Promise; + getWorkspaceSessionAsync(): Promise; // (undocumented) readonly paths: IDaemonPaths; static startAsync(options: IRushDaemonHostOptions): Promise; @@ -85,6 +250,47 @@ export class RushDaemonHost { // @beta export function serveRushDaemonAsync(options: IRushDaemonServeOptions): Promise; +// @beta +export class WorkspaceEngineComponentFactory { + constructor(options: IWorkspaceEngineComponentFactoryOptions); + // (undocumented) + readonly createAsync: CreateWorkspaceSessionComponentsAsync; + // (undocumented) + readonly shape: IWorkspaceEngineShape; +} + +// @beta +export class WorkspaceInvalidationTracker { + acknowledgeThrough(sequence: number): void; + getSnapshot(): IWorkspaceInvalidationSnapshot; + invalidate(changedPath?: string): void; + markWatcherUnhealthy(): void; +} + +// @beta +export class WorkspaceSession implements IWorkspaceSession { + [Symbol.asyncDispose](): Promise; + static createAsync(options: IWorkspaceSessionOptions): Promise; + // (undocumented) + get engineShape(): IWorkspaceEngineShape | undefined; + // (undocumented) + get inputsSnapshot(): IInputsSnapshot | undefined; + // (undocumented) + readonly invalidations: WorkspaceInvalidationTracker; + // (undocumented) + readonly metadata: IWorkspaceSessionMetadata; + // (undocumented) + readonly operationGraph: IOperationGraph | undefined; + reconcileInvalidationsAsync(): Promise; + // (undocumented) + readonly rushConfiguration: RushConfiguration; + // (undocumented) + readonly rushSession: RushSession | undefined; +} + +// @beta +export type WorkspaceSessionFactory = (options: IWorkspaceSessionOptions) => Promise; + // (No @packageDocumentation comment for this package) ``` diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index 3e895e2ee2..57a2e6fc8d 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -1,8 +1,24 @@ # @rushstack/rush-daemon The long-lived Rush workspace daemon host, including workspace-keyed listener bootstrap, -protocol handshake and liveness control, and explicit serve/shutdown lifecycle APIs. +protocol handshake and liveness control, a warm `WorkspaceSession`, and explicit +serve/shutdown lifecycle APIs. The package provides an opt-in `rushd` executable. Run it from a Rush workspace to start the host for the nearest `rush.json`; it does not change the default behavior of `rush`, `rushx`, or `rush-pnpm`. + +The host loads `RushConfiguration` once before signaling readiness and keeps a headless file watcher +active for the daemon lifetime. Its invalidation tracker retains changes while no clients are +connected so a later request can reconcile them. The tracker starts with a conservative unknown +invalidation covering session startup, and excessive distinct paths are compacted into the same +full-workspace signal. + +`WorkspaceEngineComponentFactory` provides the opt-in seam for a command integration to supply a real +all-project operation graph, its `RushSession`, and a refreshable inputs snapshot. The integration must +declare the complete phase and plugin shape because Rush plugins can currently vary that shape by command. +The factory validates graph ownership, serializes retained invalidation reconciliation, and maps path-specific +changes through the integration. The engine owner must supply one deterministic async disposer because +`IOperationGraph` does not yet expose an operation that both stops the lifetime and awaits runner cleanup. +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. diff --git a/libraries/rush-daemon/package.json b/libraries/rush-daemon/package.json index f035489be6..5c4c1c33cf 100644 --- a/libraries/rush-daemon/package.json +++ b/libraries/rush-daemon/package.json @@ -45,6 +45,7 @@ "_phase:test": "heft run --only test -- --clean" }, "dependencies": { + "@microsoft/rush-lib": "workspace:*", "@rushstack/node-core-library": "workspace:*", "@rushstack/rush-daemon-protocol": "workspace:*", "@rushstack/rush-daemon-transport": "workspace:*" diff --git a/libraries/rush-daemon/src/RushDaemonHost.ts b/libraries/rush-daemon/src/RushDaemonHost.ts index f3a9cd4251..15cc941fc3 100644 --- a/libraries/rush-daemon/src/RushDaemonHost.ts +++ b/libraries/rush-daemon/src/RushDaemonHost.ts @@ -15,6 +15,9 @@ import type { } from '@rushstack/rush-daemon-transport'; import { DaemonControlSession } from './DaemonControlSession'; +import { WorkspaceSession } from './WorkspaceSession'; +import type { IWorkspaceSession, WorkspaceSessionFactory } from './WorkspaceSession'; +import { WorkspaceSessionProvider } from './WorkspaceSessionProvider'; /** * Options for starting one workspace daemon host. @@ -22,6 +25,8 @@ import { DaemonControlSession } from './DaemonControlSession'; * @beta */ export interface IRushDaemonHostOptions { + /** Overrides workspace session construction for engine integration or testing. */ + readonly createWorkspaceSessionAsync?: WorkspaceSessionFactory; /** The daemon implementation version reported by `pong`. */ readonly daemonVersion: string; /** Reports connection-level failures. */ @@ -42,6 +47,7 @@ export interface IRushDaemonHostOptions { export class RushDaemonHost { private readonly _listener: DaemonFrameListener; private readonly _sessions: Set; + private readonly _workspaceSessionProvider: WorkspaceSessionProvider; private readonly _lifecycle: { closing: boolean }; public readonly paths: IDaemonPaths; private _closePromise: Promise | undefined; @@ -50,12 +56,14 @@ export class RushDaemonHost { listener: DaemonFrameListener, paths: IDaemonPaths, sessions: Set, - lifecycle: { closing: boolean } + lifecycle: { closing: boolean }, + workspaceSessionProvider: WorkspaceSessionProvider ) { this._listener = listener; this.paths = paths; this._sessions = sessions; this._lifecycle = lifecycle; + this._workspaceSessionProvider = workspaceSessionProvider; } /** Resolves only after the transport is bound and its lockfile has been written. */ @@ -69,29 +77,62 @@ export class RushDaemonHost { const paths: IDaemonPaths = resolveDaemonPathsFromProcess(workspaceKey); const sessions: Set = new Set(); const lifecycle: { closing: boolean } = { closing: false }; + const workspaceSessionProvider: WorkspaceSessionProvider = new WorkspaceSessionProvider( + options.createWorkspaceSessionAsync ?? WorkspaceSession.createAsync, + { + onError: options.onError, + repoRoot: canonicalRepoRoot, + rushVersion: options.rushVersion + } + ); const startedAtMs: number = Date.now(); - const listener: DaemonFrameListener = await DaemonFrameListener.listenAsync(paths, { - protocolVersion: DAEMON_PROTOCOL_VERSION, - startedAt: new Date(startedAtMs).toISOString(), - onConnection: (connection: DaemonFrameConnection) => { - const session: DaemonControlSession = new DaemonControlSession(connection, { - daemonVersion: options.daemonVersion, - startedAtMs, - onClosed: (closedSession: DaemonControlSession, error: Error | undefined) => { - sessions.delete(closedSession); - if (error) { - options.onError?.(error); - } - }, - onError: (error: Error) => options.onError?.(error) - }); - sessions.add(session); - if (lifecycle.closing) { - void session.closeAsync(); + await workspaceSessionProvider.getSessionAsync(); + let listener: DaemonFrameListener; + try { + listener = await DaemonFrameListener.listenAsync(paths, { + protocolVersion: DAEMON_PROTOCOL_VERSION, + startedAt: new Date(startedAtMs).toISOString(), + onConnection: (connection: DaemonFrameConnection) => { + const session: DaemonControlSession = new DaemonControlSession(connection, { + daemonVersion: options.daemonVersion, + startedAtMs, + onClosed: (closedSession: DaemonControlSession, error: Error | undefined) => { + sessions.delete(closedSession); + if (error) { + options.onError?.(error); + } + }, + onError: (error: Error) => options.onError?.(error) + }); + sessions.add(session); + if (lifecycle.closing) { + void session.closeAsync(); + } } + }); + } catch (error) { + try { + await workspaceSessionProvider[Symbol.asyncDispose](); + } catch (cleanupError) { + throw new AggregateError( + [error, cleanupError], + 'Failed to bind the daemon listener and dispose its workspace session.' + ); } - }); - return new RushDaemonHost(listener, paths, sessions, lifecycle); + throw error; + } + return new RushDaemonHost( + listener, + paths, + sessions, + lifecycle, + workspaceSessionProvider + ); + } + + /** Returns the single warm workspace session owned by this host. */ + public getWorkspaceSessionAsync(): Promise { + return this._workspaceSessionProvider.getSessionAsync(); } /** Closes active connections, stops listening, and removes transport artifacts. */ @@ -102,7 +143,29 @@ export class RushDaemonHost { private async _closeOnceAsync(): Promise { this._lifecycle.closing = true; - await Promise.all(Array.from(this._sessions, (session: DaemonControlSession) => session.closeAsync())); - await this._listener.closeAsync(); + const errors: unknown[] = []; + try { + await Promise.all( + Array.from(this._sessions, (session: DaemonControlSession) => session.closeAsync()) + ); + } catch (error) { + errors.push(error); + } + try { + await this._listener.closeAsync(); + } catch (error) { + errors.push(error); + } + try { + await this._workspaceSessionProvider[Symbol.asyncDispose](); + } catch (error) { + errors.push(error); + } + + if (errors.length === 1) { + throw errors[0]; + } else if (errors.length > 1) { + throw new AggregateError(errors, 'Failed to close Rush daemon host resources.'); + } } } diff --git a/libraries/rush-daemon/src/WorkspaceEngineComponentFactory.ts b/libraries/rush-daemon/src/WorkspaceEngineComponentFactory.ts new file mode 100644 index 0000000000..b9604c62d0 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceEngineComponentFactory.ts @@ -0,0 +1,386 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + GetInputsSnapshotAsyncFn, + IInputsSnapshot, + IOperationGraph, + Operation, + RushConfiguration, + RushConfigurationProject, + RushSession +} from '@microsoft/rush-lib'; + +import type { + CreateWorkspaceSessionComponentsAsync, + ICreateWorkspaceSessionComponentsOptions, + IWorkspaceSessionComponents +} from './WorkspaceSession'; +import type { + IWorkspaceInvalidationSnapshot, + WorkspaceInvalidationTracker +} from './WorkspaceInvalidationTracker'; + +const INVALIDATION_REASON: string = 'workspace-inputs-changed'; + +/** + * The command-dependent phase and plugin shape used to construct a reusable engine graph. + * + * @remarks + * The complete shape must be supplied by the integration that owns command/plugin initialization. + * Rush does not currently expose a command-independent "all phases and plugins" shape. + * + * @beta + */ +export interface IWorkspaceEngineShape { + readonly phaseNames: ReadonlyArray; + readonly pluginNames: ReadonlyArray; +} + +/** + * Context for constructing real reusable Rush engine components. + * + * @beta + */ +export interface ICreateWorkspaceEngineComponentsOptions extends IWorkspaceEngineShape { + /** Every project in the loaded Rush configuration. */ + readonly projectSelection: ReadonlySet; + readonly rushConfiguration: RushConfiguration; +} + +/** + * Real Rush engine state constructed by the command/plugin integration. + * + * @beta + */ +export interface IWorkspaceEngineComponents extends AsyncDisposable { + /** + * Stops the graph lifetime and awaits all iteration, runner, plugin, and snapshot resources. + * + * @remarks + * The engine owner must implement this because the public `IOperationGraph` API does not + * currently expose one deterministic shutdown operation. + */ + [Symbol.asyncDispose](): Promise; + readonly getInputsSnapshotAsync: GetInputsSnapshotAsyncFn; + readonly inputsSnapshot: IInputsSnapshot; + readonly operationGraph: IOperationGraph; + readonly rushSession: RushSession; +} + +/** + * Constructs reusable Rush engine state for an explicit graph shape. + * + * @beta + */ +export type CreateWorkspaceEngineComponentsAsync = ( + options: ICreateWorkspaceEngineComponentsOptions +) => Promise; + +/** + * Context for mapping watcher paths onto operations in a constructed graph. + * + * @beta + */ +export interface IMapWorkspaceInvalidationsOptions { + readonly changedPaths: ReadonlyArray; + readonly currentInputsSnapshot: IInputsSnapshot; + readonly nextInputsSnapshot: IInputsSnapshot; + readonly operationGraph: IOperationGraph; +} + +/** + * Maps path-specific watcher invalidations onto operations in the reusable graph. + * + * @remarks + * Unknown changes and unhealthy watcher state bypass this callback and invalidate the full graph. + * + * @beta + */ +export type MapWorkspaceInvalidationsToOperationsAsync = ( + options: IMapWorkspaceInvalidationsOptions +) => Promise>; + +/** + * The result of reconciling retained watcher invalidations with the engine snapshot. + * + * @beta + */ +export interface IWorkspaceInvalidationReconciliation { + readonly inputsSnapshot: IInputsSnapshot; + readonly invalidatedOperationCount: number; + readonly isFullInvalidation: boolean; + readonly sequence: number; +} + +/** + * Options for {@link WorkspaceEngineComponentFactory}. + * + * @beta + */ +export interface IWorkspaceEngineComponentFactoryOptions { + readonly createEngineComponentsAsync: CreateWorkspaceEngineComponentsAsync; + readonly mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; + readonly shape: IWorkspaceEngineShape; +} + +interface IWorkspaceEngineLifecycleOptions { + readonly components: IWorkspaceEngineComponents; + readonly invalidations: WorkspaceInvalidationTracker; + readonly mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; +} + +class WorkspaceEngineLifecycle { + readonly #components: IWorkspaceEngineComponents; + readonly #invalidations: WorkspaceInvalidationTracker; + readonly #mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; + #currentInputsSnapshot: IInputsSnapshot; + #disposePromise: Promise | undefined; + #isDisposing: boolean = false; + #reconciliationTail: Promise = Promise.resolve(); + #requiresFullInvalidation: boolean = false; + + public constructor(options: IWorkspaceEngineLifecycleOptions) { + this.#components = options.components; + this.#currentInputsSnapshot = options.components.inputsSnapshot; + this.#invalidations = options.invalidations; + this.#mapInvalidationsToOperationsAsync = options.mapInvalidationsToOperationsAsync; + } + + public get inputsSnapshot(): IInputsSnapshot { + return this.#currentInputsSnapshot; + } + + public reconcileInvalidationsAsync(): Promise { + if (this.#isDisposing) { + return Promise.reject(new Error('The workspace engine is being disposed.')); + } + + const reconciliationPromise: Promise = + this.#reconciliationTail.then(() => this.#reconcileOnceAsync()); + this.#reconciliationTail = reconciliationPromise.then( + () => undefined, + () => undefined + ); + return reconciliationPromise; + } + + public [Symbol.asyncDispose](): Promise { + this.#isDisposing = true; + this.#disposePromise ??= this.#disposeOnceAsync(); + return this.#disposePromise; + } + + async #reconcileOnceAsync(): Promise { + const invalidationSnapshot: IWorkspaceInvalidationSnapshot = this.#invalidations.getSnapshot(); + const isFullInvalidation: boolean = + this.#requiresFullInvalidation || + invalidationSnapshot.hasUnknownChanges || + !invalidationSnapshot.isWatcherHealthy; + if (!isFullInvalidation && invalidationSnapshot.changedPaths.length === 0) { + return { + inputsSnapshot: this.#currentInputsSnapshot, + invalidatedOperationCount: 0, + isFullInvalidation: false, + sequence: invalidationSnapshot.sequence + }; + } + + const nextInputsSnapshot: IInputsSnapshot | undefined = + await this.#components.getInputsSnapshotAsync(); + if (!nextInputsSnapshot) { + throw new Error('Rush could not capture the next workspace inputs snapshot.'); + } + + const operationGraph: IOperationGraph = this.#components.operationGraph; + let invalidatedOperationCount: number; + if (isFullInvalidation) { + operationGraph.invalidateOperations(undefined, INVALIDATION_REASON); + invalidatedOperationCount = operationGraph.operations.size; + } else { + const mappedOperations: Iterable = + await this.#mapInvalidationsToOperationsAsync({ + changedPaths: invalidationSnapshot.changedPaths, + currentInputsSnapshot: this.#currentInputsSnapshot, + nextInputsSnapshot, + operationGraph + }); + const invalidatedOperations: ReadonlySet = validateMappedOperations( + mappedOperations, + operationGraph + ); + operationGraph.invalidateOperations(invalidatedOperations, INVALIDATION_REASON); + invalidatedOperationCount = invalidatedOperations.size; + } + + this.#currentInputsSnapshot = nextInputsSnapshot; + this.#invalidations.acknowledgeThrough(invalidationSnapshot.sequence); + this.#requiresFullInvalidation = + this.#invalidations.getSnapshot().sequence > invalidationSnapshot.sequence; + return { + inputsSnapshot: nextInputsSnapshot, + invalidatedOperationCount, + isFullInvalidation, + sequence: invalidationSnapshot.sequence + }; + } + + async #disposeOnceAsync(): Promise { + await this.#reconciliationTail; + await this.#components[Symbol.asyncDispose](); + } +} + +/** + * Adapts an explicitly shaped, all-project Rush engine into warm workspace session components. + * + * @remarks + * The factory deliberately does not construct phased hooks or load plugins itself. Those choices remain + * command-specific in Rush today, so their owner must provide the complete shape and construction callback. + * + * @beta + */ +export class WorkspaceEngineComponentFactory { + readonly #createEngineComponentsAsync: CreateWorkspaceEngineComponentsAsync; + readonly #mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; + + public readonly createAsync: CreateWorkspaceSessionComponentsAsync; + public readonly shape: IWorkspaceEngineShape; + + public constructor(options: IWorkspaceEngineComponentFactoryOptions) { + this.#createEngineComponentsAsync = options.createEngineComponentsAsync; + this.#mapInvalidationsToOperationsAsync = options.mapInvalidationsToOperationsAsync; + this.shape = normalizeShape(options.shape); + this.createAsync = (createOptions: ICreateWorkspaceSessionComponentsOptions) => + this.#createAsync(createOptions); + } + + async #createAsync( + options: ICreateWorkspaceSessionComponentsOptions + ): Promise { + const projects: ReadonlySet = new Set(options.rushConfiguration.projects); + const components: IWorkspaceEngineComponents = await this.#createEngineComponentsAsync({ + phaseNames: this.shape.phaseNames, + pluginNames: this.shape.pluginNames, + projectSelection: projects, + rushConfiguration: options.rushConfiguration + }); + + try { + validateComponents(components, options.rushConfiguration, this.shape); + } catch (error) { + await disposeAfterInitializationFailureAsync(components, error); + } + + const lifecycle: WorkspaceEngineLifecycle = new WorkspaceEngineLifecycle({ + components, + invalidations: options.invalidations, + mapInvalidationsToOperationsAsync: this.#mapInvalidationsToOperationsAsync + }); + return { + [Symbol.asyncDispose]: () => lifecycle[Symbol.asyncDispose](), + engineShape: this.shape, + get inputsSnapshot(): IInputsSnapshot { + return lifecycle.inputsSnapshot; + }, + operationGraph: components.operationGraph, + reconcileInvalidationsAsync: () => lifecycle.reconcileInvalidationsAsync(), + rushSession: components.rushSession + }; + } +} + +function normalizeShape(shape: IWorkspaceEngineShape): IWorkspaceEngineShape { + return Object.freeze({ + phaseNames: normalizeNames(shape.phaseNames, 'phase', true), + pluginNames: normalizeNames(shape.pluginNames, 'plugin', false) + }); +} + +function normalizeNames( + names: ReadonlyArray, + kind: string, + requireNonempty: boolean +): ReadonlyArray { + if (requireNonempty && names.length === 0) { + throw new Error(`The workspace engine ${kind} shape must not be empty.`); + } + + const normalizedNames: Set = new Set(); + for (const name of names) { + if (name.length === 0 || name.trim() !== name) { + throw new Error(`Invalid workspace engine ${kind} name: "${name}".`); + } + if (normalizedNames.has(name)) { + throw new Error(`Duplicate workspace engine ${kind} name: "${name}".`); + } + normalizedNames.add(name); + } + return Object.freeze(Array.from(normalizedNames).sort()); +} + +function validateComponents( + components: IWorkspaceEngineComponents, + rushConfiguration: RushConfiguration, + shape: IWorkspaceEngineShape +): void { + const operations: ReadonlySet = components.operationGraph.operations; + if (operations.size === 0) { + throw new Error('The reusable workspace operation graph must not be empty.'); + } + + const configuredProjects: ReadonlySet = new Set( + rushConfiguration.projects + ); + const representedProjects: Set = new Set(); + const phaseNames: ReadonlySet = new Set(shape.phaseNames); + for (const operation of operations) { + if (!configuredProjects.has(operation.associatedProject)) { + throw new Error( + `Operation "${operation.associatedPhase.name}" uses a project outside the loaded Rush configuration.` + ); + } + if (!phaseNames.has(operation.associatedPhase.name)) { + throw new Error( + `Operation phase "${operation.associatedPhase.name}" is not declared in the workspace engine shape.` + ); + } + representedProjects.add(operation.associatedProject); + } + for (const project of configuredProjects) { + if (!representedProjects.has(project)) { + throw new Error( + `The reusable workspace operation graph does not represent project "${project.packageName}".` + ); + } + } +} + +function validateMappedOperations( + operations: Iterable, + operationGraph: IOperationGraph +): ReadonlySet { + const validatedOperations: Set = new Set(); + for (const operation of operations) { + if (!operationGraph.operations.has(operation)) { + throw new Error('The workspace invalidation mapper returned an operation outside the graph.'); + } + validatedOperations.add(operation); + } + return validatedOperations; +} + +async function disposeAfterInitializationFailureAsync( + components: IWorkspaceEngineComponents, + initializationError: unknown +): Promise { + try { + await components[Symbol.asyncDispose](); + } catch (cleanupError) { + throw new AggregateError( + [initializationError, cleanupError], + 'Failed to validate and clean up reusable workspace engine components.' + ); + } + throw initializationError; +} diff --git a/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts new file mode 100644 index 0000000000..ee67d5afa1 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts @@ -0,0 +1,98 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** + * A point-in-time view of workspace changes that have not yet been reconciled. + * + * @beta + */ +export interface IWorkspaceInvalidationSnapshot { + /** Paths reported by the watcher, sorted for deterministic consumption. */ + readonly changedPaths: ReadonlyArray; + /** True when the watcher reported a change without a path or encountered a watcher error. */ + readonly hasUnknownChanges: boolean; + /** False after a watcher error makes subsequent change detection unreliable. */ + readonly isWatcherHealthy: boolean; + /** The latest invalidation sequence included in this snapshot. */ + readonly sequence: number; +} + +const MAX_TRACKED_CHANGED_PATHS: number = 10_000; + +/** + * Retains workspace invalidations until a future request explicitly acknowledges them. + * + * @beta + */ +export class WorkspaceInvalidationTracker { + readonly #sequenceByPath: Map = new Map(); + #latestSequence: number = 0; + #unknownChangeSequence: number | undefined; + #watcherHealthy: boolean = true; + + /** Records a path-specific or unknown workspace change. */ + public invalidate(changedPath?: string): void { + const sequence: number = ++this.#latestSequence; + if (changedPath === undefined || this.#unknownChangeSequence !== undefined) { + this.#unknownChangeSequence = sequence; + return; + } + + if ( + !this.#sequenceByPath.has(changedPath) && + this.#sequenceByPath.size >= MAX_TRACKED_CHANGED_PATHS + ) { + this.#sequenceByPath.clear(); + this.#unknownChangeSequence = sequence; + return; + } + + this.#sequenceByPath.set(changedPath, sequence); + } + + /** + * Permanently marks the current watcher as unhealthy. + * + * Unknown invalidation remains pending so consumers cannot mistake the workspace for clean. + */ + public markWatcherUnhealthy(): void { + if (this.#watcherHealthy) { + this.#watcherHealthy = false; + this.invalidate(); + } + } + + /** Returns all changes that have not been acknowledged. */ + public getSnapshot(): IWorkspaceInvalidationSnapshot { + return { + changedPaths: Array.from(this.#sequenceByPath.keys()).sort(), + hasUnknownChanges: this.#unknownChangeSequence !== undefined, + isWatcherHealthy: this.#watcherHealthy, + sequence: this.#latestSequence + }; + } + + /** + * Acknowledges changes through a previously observed sequence. + * + * Changes that arrive after that sequence remain pending, including repeated changes to the same path. + */ + public acknowledgeThrough(sequence: number): void { + if (!Number.isSafeInteger(sequence) || sequence < 0 || sequence > this.#latestSequence) { + throw new RangeError(`Invalid workspace invalidation sequence: ${sequence}`); + } + + for (const [changedPath, pathSequence] of this.#sequenceByPath) { + if (pathSequence <= sequence) { + this.#sequenceByPath.delete(changedPath); + } + } + if ( + this.#watcherHealthy && + this.#unknownChangeSequence !== undefined && + this.#unknownChangeSequence <= sequence + ) { + this.#unknownChangeSequence = undefined; + } + } +} diff --git a/libraries/rush-daemon/src/WorkspaceSession.ts b/libraries/rush-daemon/src/WorkspaceSession.ts new file mode 100644 index 0000000000..606c58ebd2 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSession.ts @@ -0,0 +1,299 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as path from 'node:path'; + +import { RushConfiguration } from '@microsoft/rush-lib'; +import type { + IInputsSnapshot, + IOperationGraph, + RushSession +} from '@microsoft/rush-lib'; + +import { WorkspaceInvalidationTracker } from './WorkspaceInvalidationTracker'; +import { WorkspaceSessionFileWatcher } from './WorkspaceSessionFileWatcher'; +import type { + IWorkspaceEngineShape, + IWorkspaceInvalidationReconciliation +} from './WorkspaceEngineComponentFactory'; + +/** + * Stable identity loaded once for a warm workspace session. + * + * @beta + */ +export interface IWorkspaceSessionMetadata { + readonly projectCount: number; + readonly projectNames: ReadonlyArray; + readonly repoRoot: string; + readonly rushJsonFile: string; + readonly rushVersion: string; +} + +/** + * A headless watcher that reports changes independently of connected clients. + * + * @beta + */ +export interface IWorkspaceInvalidationWatcher extends AsyncDisposable { + startAsync(onInvalidation: (changedPath?: string) => void): Promise; +} + +/** + * Optional engine state supplied when reusable graph construction is available. + * + * @remarks + * The default session intentionally leaves the graph, plugin session, and inputs snapshot uninitialized. + * Their existing construction is command-specific and remains blocked on the reusable runner lifetime work. + * + * @beta + */ +export interface IWorkspaceSessionComponents extends AsyncDisposable { + readonly engineShape?: IWorkspaceEngineShape; + readonly inputsSnapshot?: IInputsSnapshot; + readonly operationGraph?: IOperationGraph; + /** + * An injected watcher owned by this component bundle. + * + * @remarks + * When provided, the component bundle's async disposer must dispose the watcher. + * `WorkspaceSession` directly disposes only the default watcher that it creates itself. + */ + readonly projectWatcher?: IWorkspaceInvalidationWatcher; + readonly reconcileInvalidationsAsync?: () => Promise; + readonly rushSession?: RushSession; +} + +/** + * Context for constructing optional reusable workspace engine components. + * + * @beta + */ +export interface ICreateWorkspaceSessionComponentsOptions { + readonly invalidations: WorkspaceInvalidationTracker; + readonly onError?: (error: Error) => void; + readonly rushConfiguration: RushConfiguration; +} + +/** + * Constructs optional reusable graph, plugin, snapshot, and watcher state. + * + * @beta + */ +export type CreateWorkspaceSessionComponentsAsync = ( + options: ICreateWorkspaceSessionComponentsOptions +) => Promise; + +/** + * Options for initializing a workspace session. + * + * @beta + */ +export interface IWorkspaceSessionOptions { + readonly createComponentsAsync?: CreateWorkspaceSessionComponentsAsync; + readonly onError?: (error: Error) => void; + readonly repoRoot: string; + readonly rushVersion: string; +} + +/** + * The reusable state owned by one daemon lifecycle. + * + * @beta + */ +export interface IWorkspaceSession extends AsyncDisposable { + readonly engineShape: IWorkspaceEngineShape | undefined; + readonly inputsSnapshot: IInputsSnapshot | undefined; + readonly invalidations: WorkspaceInvalidationTracker; + readonly metadata: IWorkspaceSessionMetadata; + readonly operationGraph: IOperationGraph | undefined; + readonly rushConfiguration: RushConfiguration; + readonly rushSession: RushSession | undefined; + reconcileInvalidationsAsync(): Promise; +} + +/** + * Factory used by the daemon host to initialize its workspace session. + * + * @beta + */ +export type WorkspaceSessionFactory = (options: IWorkspaceSessionOptions) => Promise; + +const EMPTY_WORKSPACE_SESSION_COMPONENTS: IWorkspaceSessionComponents = { + [Symbol.asyncDispose]: () => Promise.resolve() +}; + +/** + * A warm workspace session with client-independent invalidation tracking. + * + * @beta + */ +export class WorkspaceSession implements IWorkspaceSession { + readonly #components: IWorkspaceSessionComponents; + readonly #sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined; + #disposePromise: Promise | undefined; + #inputsSnapshot: IInputsSnapshot | undefined; + #isDisposing: boolean = false; + + public readonly invalidations: WorkspaceInvalidationTracker; + public readonly metadata: IWorkspaceSessionMetadata; + public readonly operationGraph: IOperationGraph | undefined; + public readonly rushConfiguration: RushConfiguration; + public readonly rushSession: RushSession | undefined; + + private constructor( + rushConfiguration: RushConfiguration, + metadata: IWorkspaceSessionMetadata, + invalidations: WorkspaceInvalidationTracker, + components: IWorkspaceSessionComponents, + sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined + ) { + this.rushConfiguration = rushConfiguration; + this.metadata = metadata; + this.invalidations = invalidations; + this.#components = components; + this.#sessionOwnedProjectWatcher = sessionOwnedProjectWatcher; + this.#inputsSnapshot = components.inputsSnapshot; + this.operationGraph = components.operationGraph; + this.rushSession = components.rushSession; + } + + public get engineShape(): IWorkspaceEngineShape | undefined { + return this.#components.engineShape; + } + + public get inputsSnapshot(): IInputsSnapshot | undefined { + return this.#inputsSnapshot; + } + + /** Loads workspace identity, creates reusable components, and starts headless invalidation tracking. */ + public static async createAsync(options: IWorkspaceSessionOptions): Promise { + const rushConfiguration: RushConfiguration = RushConfiguration.loadFromConfigurationFile( + path.join(options.repoRoot, 'rush.json') + ); + const canonicalRepoRoot: string = path.resolve(options.repoRoot); + if (path.resolve(rushConfiguration.rushJsonFolder) !== canonicalRepoRoot) { + throw new Error(`Rush configuration resolved outside the daemon workspace: ${options.repoRoot}`); + } + + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + const components: IWorkspaceSessionComponents = + (await options.createComponentsAsync?.({ + invalidations, + onError: options.onError, + rushConfiguration + })) ?? EMPTY_WORKSPACE_SESSION_COMPONENTS; + let projectWatcher: IWorkspaceInvalidationWatcher | undefined = components.projectWatcher; + let sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined; + try { + const metadata: IWorkspaceSessionMetadata = createMetadata( + rushConfiguration, + options.rushVersion + ); + if (!projectWatcher) { + projectWatcher = new WorkspaceSessionFileWatcher({ + onError: (error: Error) => { + invalidations.markWatcherUnhealthy(); + options.onError?.(error); + }, + rushConfiguration + }); + sessionOwnedProjectWatcher = projectWatcher; + } + const session: WorkspaceSession = new WorkspaceSession( + rushConfiguration, + metadata, + invalidations, + components, + sessionOwnedProjectWatcher + ); + await projectWatcher.startAsync((changedPath: string | undefined) => + invalidations.invalidate(changedPath) + ); + // Changes before the watcher registered its callbacks cannot be observed path-by-path. + invalidations.invalidate(); + return session; + } catch (error) { + const cleanupErrors: unknown[] = []; + try { + await sessionOwnedProjectWatcher?.[Symbol.asyncDispose](); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + try { + await components[Symbol.asyncDispose](); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + if (cleanupErrors.length > 0) { + throw new AggregateError( + [error, ...cleanupErrors], + 'Failed to initialize and clean up the workspace session.' + ); + } + throw error; + } + } + + /** Stops invalidation tracking and disposes injected engine resources. */ + public [Symbol.asyncDispose](): Promise { + this.#isDisposing = true; + this.#disposePromise ??= this.#disposeOnceAsync(); + return this.#disposePromise; + } + + /** Reconciles retained watcher changes with injected reusable engine state, when configured. */ + public async reconcileInvalidationsAsync(): Promise { + if (this.#isDisposing) { + throw new Error('The workspace session is being disposed.'); + } + if (!this.#components.reconcileInvalidationsAsync) { + return undefined; + } + const result: IWorkspaceInvalidationReconciliation = + await this.#components.reconcileInvalidationsAsync(); + this.#inputsSnapshot = result.inputsSnapshot; + return result; + } + + async #disposeOnceAsync(): Promise { + let watcherError: unknown; + try { + await this.#sessionOwnedProjectWatcher?.[Symbol.asyncDispose](); + } catch (error) { + watcherError = error; + } + + try { + await this.#components[Symbol.asyncDispose](); + } catch (componentError) { + if (watcherError !== undefined) { + throw new AggregateError( + [watcherError, componentError], + 'Failed to dispose workspace session resources.' + ); + } + throw componentError; + } + if (watcherError !== undefined) { + throw watcherError; + } + } +} + +function createMetadata( + rushConfiguration: RushConfiguration, + rushVersion: string +): IWorkspaceSessionMetadata { + const projectNames: string[] = Array.from( + rushConfiguration.projects, + (project) => project.packageName + ).sort(); + return { + projectCount: projectNames.length, + projectNames, + repoRoot: rushConfiguration.rushJsonFolder, + rushJsonFile: rushConfiguration.rushJsonFile, + rushVersion + }; +} diff --git a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts new file mode 100644 index 0000000000..acee521b21 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts @@ -0,0 +1,121 @@ +// 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 { once } from 'node:events'; + +import type { RushConfiguration } from '@microsoft/rush-lib'; + +import type { IWorkspaceInvalidationWatcher } from './WorkspaceSession'; + +export interface IWorkspaceSessionFileWatcherOptions { + readonly onError?: (error: Error) => void; + readonly rushConfiguration: RushConfiguration; + readonly watchFactory?: WorkspaceWatchFactory; +} + +interface IWatchPath { + readonly folderPath: string; + readonly recursive: boolean; +} + +type WorkspaceWatchFactory = ( + folderPath: string, + options: { encoding: 'utf8'; recursive: boolean }, + listener: fs.WatchListener +) => fs.FSWatcher; + +const PATH_SEGMENT_SEPARATOR_REGEXP: RegExp = /[\\/]/; + +export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatcher { + readonly #onError: ((error: Error) => void) | undefined; + readonly #watchFactory: WorkspaceWatchFactory; + readonly #watchPaths: ReadonlyArray; + readonly #watchers: Set = new Set(); + #onInvalidation: ((changedPath?: string) => void) | undefined; + #disposed: boolean = false; + + public constructor(options: IWorkspaceSessionFileWatcherOptions) { + this.#onError = options.onError; + this.#watchFactory = options.watchFactory ?? fs.watch; + this.#watchPaths = getWatchPaths(options.rushConfiguration); + } + + public async startAsync(onInvalidation: (changedPath?: string) => void): Promise { + if (this.#disposed) { + throw new Error('The workspace watcher has already been disposed.'); + } + if (this.#onInvalidation) { + throw new Error('The workspace watcher has already been started.'); + } + + this.#onInvalidation = onInvalidation; + for (const watchPath of this.#watchPaths) { + this.#watchers.add(this.#createWatcher(watchPath)); + } + } + + public async [Symbol.asyncDispose](): Promise { + if (this.#disposed) { + return; + } + this.#disposed = true; + const closePromises: Promise[] = []; + for (const watcher of this.#watchers) { + closePromises.push(once(watcher, 'close')); + watcher.close(); + } + await Promise.all(closePromises); + this.#watchers.clear(); + this.#onInvalidation = undefined; + } + + #createWatcher(watchPath: IWatchPath): fs.FSWatcher { + const watcher: fs.FSWatcher = this.#watchFactory( + watchPath.folderPath, + { encoding: 'utf8', recursive: watchPath.recursive }, + (eventType: string, filename: string | null) => { + void eventType; + const changedFilename: string | undefined = filename ?? undefined; + if (!isIgnoredPath(changedFilename)) { + this.#onInvalidation?.( + changedFilename === undefined + ? undefined + : path.resolve(watchPath.folderPath, changedFilename) + ); + } + } + ); + watcher.on('error', (error: Error) => { + this.#onInvalidation?.(); + this.#onError?.(error); + }); + watcher.once('close', () => this.#watchers.delete(watcher)); + watcher.unref(); + return watcher; + } +} + +function getWatchPaths(rushConfiguration: RushConfiguration): ReadonlyArray { + const recursiveFolders: Set = new Set([rushConfiguration.commonRushConfigFolder]); + for (const subspace of rushConfiguration.subspaces) { + recursiveFolders.add(subspace.getSubspaceConfigFolderPath()); + } + for (const project of rushConfiguration.projects) { + recursiveFolders.add(project.projectFolder); + } + return [ + { folderPath: rushConfiguration.rushJsonFolder, recursive: false }, + ...Array.from(recursiveFolders, (folderPath: string) => ({ folderPath, recursive: true })) + ]; +} + +function isIgnoredPath(filename: string | undefined): boolean { + if (filename === undefined) { + return false; + } + return filename + .split(PATH_SEGMENT_SEPARATOR_REGEXP) + .some((segment: string) => segment === '.git' || segment === 'node_modules'); +} diff --git a/libraries/rush-daemon/src/WorkspaceSessionProvider.ts b/libraries/rush-daemon/src/WorkspaceSessionProvider.ts new file mode 100644 index 0000000000..85ba8eba51 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSessionProvider.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. + +import type { + IWorkspaceSession, + IWorkspaceSessionOptions, + WorkspaceSessionFactory +} from './WorkspaceSession'; + +export class WorkspaceSessionProvider implements AsyncDisposable { + readonly #factory: WorkspaceSessionFactory; + readonly #options: IWorkspaceSessionOptions; + #disposePromise: Promise | undefined; + #initializationDisposalPromise: Promise | undefined; + #initializationPromise: Promise | undefined; + #session: IWorkspaceSession | undefined; + #disposed: boolean = false; + + public constructor(factory: WorkspaceSessionFactory, options: IWorkspaceSessionOptions) { + this.#factory = factory; + this.#options = options; + } + + public getSessionAsync(): Promise { + if (this.#disposed) { + return Promise.reject(new Error('The workspace session provider has been disposed.')); + } + if (this.#session) { + return Promise.resolve(this.#session); + } + if (!this.#initializationPromise) { + const initializationPromise: Promise = Promise.resolve().then(() => + this.#initializeAsync() + ); + this.#initializationPromise = initializationPromise; + void initializationPromise.catch(() => { + if (this.#initializationPromise === initializationPromise) { + this.#initializationPromise = undefined; + } + }); + } + return this.#initializationPromise; + } + + public [Symbol.asyncDispose](): Promise { + this.#disposePromise ??= this.#disposeOnceAsync(); + return this.#disposePromise; + } + + async #disposeOnceAsync(): Promise { + this.#disposed = true; + try { + const session: IWorkspaceSession | undefined = + this.#session ?? + (await this.#initializationPromise?.then( + (initializedSession: IWorkspaceSession) => initializedSession, + () => undefined + )); + if (session) { + await session[Symbol.asyncDispose](); + } else { + await this.#initializationDisposalPromise; + } + } finally { + this.#session = undefined; + this.#initializationPromise = undefined; + this.#initializationDisposalPromise = undefined; + } + } + + async #initializeAsync(): Promise { + const session: IWorkspaceSession = await this.#factory(this.#options); + if (this.#disposed) { + this.#initializationDisposalPromise = Promise.resolve().then(() => + session[Symbol.asyncDispose]() + ); + await this.#initializationDisposalPromise; + throw new Error('The workspace session provider was disposed during initialization.'); + } + this.#session = session; + return session; + } +} diff --git a/libraries/rush-daemon/src/index.ts b/libraries/rush-daemon/src/index.ts index a9178e9749..1634b98560 100644 --- a/libraries/rush-daemon/src/index.ts +++ b/libraries/rush-daemon/src/index.ts @@ -13,3 +13,29 @@ export { } from './RequestScheduler'; export { RushDaemonHost, type IRushDaemonHostOptions } from './RushDaemonHost'; export { serveRushDaemonAsync, type IRushDaemonServeOptions } from './serveRushDaemon'; +export { + WorkspaceEngineComponentFactory, + type CreateWorkspaceEngineComponentsAsync, + type ICreateWorkspaceEngineComponentsOptions, + type IMapWorkspaceInvalidationsOptions, + type IWorkspaceEngineComponentFactoryOptions, + type IWorkspaceEngineComponents, + type IWorkspaceEngineShape, + type IWorkspaceInvalidationReconciliation, + type MapWorkspaceInvalidationsToOperationsAsync +} from './WorkspaceEngineComponentFactory'; +export { + WorkspaceSession, + type CreateWorkspaceSessionComponentsAsync, + type ICreateWorkspaceSessionComponentsOptions, + type IWorkspaceInvalidationWatcher, + type IWorkspaceSession, + type IWorkspaceSessionComponents, + type IWorkspaceSessionMetadata, + type IWorkspaceSessionOptions, + type WorkspaceSessionFactory +} from './WorkspaceSession'; +export { + WorkspaceInvalidationTracker, + type IWorkspaceInvalidationSnapshot +} from './WorkspaceInvalidationTracker'; diff --git a/libraries/rush-daemon/src/test/RushDaemonHost.test.ts b/libraries/rush-daemon/src/test/RushDaemonHost.test.ts index 768d95e10e..020d7e5ed3 100644 --- a/libraries/rush-daemon/src/test/RushDaemonHost.test.ts +++ b/libraries/rush-daemon/src/test/RushDaemonHost.test.ts @@ -17,8 +17,10 @@ import type { IDaemonFrame } from '@rushstack/rush-daemon-protocol'; import { + computeDaemonWorkspaceKey, connectDaemonAsync, - readDaemonLockfile + readDaemonLockfile, + resolveDaemonPathsFromProcess } from '@rushstack/rush-daemon-transport'; import type { DaemonFrameConnection, @@ -26,7 +28,10 @@ import type { } from '@rushstack/rush-daemon-transport'; import { RushDaemonHost } from '../RushDaemonHost'; +import type { IRushDaemonHostOptions } from '../RushDaemonHost'; import { serveRushDaemonAsync } from '../serveRushDaemon'; +import type { IWorkspaceSession } from '../WorkspaceSession'; +import { TestWorkspaceSession } from './TestWorkspaceSession'; const RUSH_VERSION: string = '5.178.1'; const DAEMON_VERSION: string = '0.1.0-test'; @@ -48,12 +53,17 @@ function createTestRepoRoot(): string { return repoRoot; } -function createHostOptions(repoRoot: string): { - daemonVersion: string; - repoRoot: string; - rushVersion: string; -} { - return { daemonVersion: DAEMON_VERSION, repoRoot, rushVersion: RUSH_VERSION }; +function createHostOptions( + repoRoot: string, + overrides: Partial = {} +): IRushDaemonHostOptions { + return { + daemonVersion: DAEMON_VERSION, + repoRoot, + rushVersion: RUSH_VERSION, + createWorkspaceSessionAsync: () => Promise.resolve(new TestWorkspaceSession(repoRoot)), + ...overrides + }; } async function exchangeControlAsync( @@ -134,16 +144,99 @@ describe(RushDaemonHost.name, () => { }); it('closes active connections and removes transport artifacts', async () => { + const disposalEvents: string[] = []; + const repoRoot: string = createTestRepoRoot(); const host: RushDaemonHost = await RushDaemonHost.startAsync( - createHostOptions(createTestRepoRoot()) + createHostOptions(repoRoot, { + createWorkspaceSessionAsync: () => + Promise.resolve( + new TestWorkspaceSession(repoRoot, () => disposalEvents.push('workspace-session')) + ) + }) ); const client: DaemonFrameConnection = await connectDaemonAsync(host.paths.socketPath); - const closed: Promise = new Promise((resolve: () => void) => client.onClosed(() => resolve())); + const closed: Promise = new Promise((resolve: () => void) => + client.onClosed(() => { + disposalEvents.push('client'); + resolve(); + }) + ); await host.closeAsync(); await closed; + expect(disposalEvents).toHaveLength(2); + expect(disposalEvents).toContain('client'); + expect(disposalEvents).toContain('workspace-session'); expect(readDaemonLockfile(host.paths.lockfilePath)).toBeUndefined(); await expect(connectDaemonAsync(host.paths.socketPath)).rejects.toMatchObject({ code: 'connectionRefused' }); }); + + it('initializes one workspace session and reuses it', async () => { + const repoRoot: string = createTestRepoRoot(); + const workspaceSession: IWorkspaceSession = new TestWorkspaceSession(repoRoot); + let factoryCalls: number = 0; + const host: RushDaemonHost = await RushDaemonHost.startAsync( + createHostOptions(repoRoot, { + createWorkspaceSessionAsync: () => { + factoryCalls++; + return Promise.resolve(workspaceSession); + } + }) + ); + try { + const [first, second] = await Promise.all([ + host.getWorkspaceSessionAsync(), + host.getWorkspaceSessionAsync() + ]); + expect(first).toBe(workspaceSession); + expect(second).toBe(workspaceSession); + expect(factoryCalls).toBe(1); + } finally { + await host.closeAsync(); + } + }); + + it('removes transport artifacts when workspace initialization fails', async () => { + const repoRoot: string = createTestRepoRoot(); + const workspaceKey: string = computeDaemonWorkspaceKey({ + canonicalRepoRoot: fs.realpathSync(repoRoot), + rushVersion: RUSH_VERSION + }); + const paths: IDaemonPaths = resolveDaemonPathsFromProcess(workspaceKey); + + await expect( + RushDaemonHost.startAsync( + createHostOptions(repoRoot, { + createWorkspaceSessionAsync: () => Promise.reject(new Error('workspace failed')) + }) + ) + ).rejects.toThrow('workspace failed'); + + expect(readDaemonLockfile(paths.lockfilePath)).toBeUndefined(); + await expect(connectDaemonAsync(paths.socketPath)).rejects.toMatchObject({ + code: 'connectionRefused' + }); + }); + + it('disposes the workspace session when listener binding fails', async () => { + const repoRoot: string = createTestRepoRoot(); + const firstHost: RushDaemonHost = await RushDaemonHost.startAsync(createHostOptions(repoRoot)); + const disposalEvents: string[] = []; + try { + await expect( + RushDaemonHost.startAsync( + createHostOptions(repoRoot, { + createWorkspaceSessionAsync: () => + Promise.resolve( + new TestWorkspaceSession(repoRoot, () => disposalEvents.push('workspace-session')) + ) + }) + ) + ).rejects.toMatchObject({ code: 'daemonAlreadyRunning' }); + expect(disposalEvents).toEqual(['workspace-session']); + } finally { + await firstHost.closeAsync(); + } + }); }); diff --git a/libraries/rush-daemon/src/test/TestWorkspaceSession.ts b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts new file mode 100644 index 0000000000..4f917d98ff --- /dev/null +++ b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts @@ -0,0 +1,57 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as path from 'node:path'; + +import { RushConfiguration } from '@microsoft/rush-lib'; +import type { + IInputsSnapshot, + IOperationGraph, + RushSession +} from '@microsoft/rush-lib'; + +import type { + IWorkspaceEngineShape, + IWorkspaceInvalidationReconciliation +} from '../WorkspaceEngineComponentFactory'; +import type { + IWorkspaceSession, + IWorkspaceSessionMetadata +} from '../WorkspaceSession'; +import { WorkspaceInvalidationTracker } from '../WorkspaceInvalidationTracker'; + +export const TEST_REPO_ROOT: string = path.resolve(__dirname, '../../../..'); +export const TEST_RUSH_CONFIGURATION: RushConfiguration = RushConfiguration.loadFromConfigurationFile( + path.join(TEST_REPO_ROOT, 'rush.json') +); + +export class TestWorkspaceSession implements IWorkspaceSession { + readonly #onDispose: (() => unknown) | undefined; + + public readonly engineShape: IWorkspaceEngineShape | undefined; + public readonly inputsSnapshot: IInputsSnapshot | undefined; + public readonly invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + public readonly metadata: IWorkspaceSessionMetadata; + public readonly operationGraph: IOperationGraph | undefined; + public readonly rushConfiguration: RushConfiguration = TEST_RUSH_CONFIGURATION; + public readonly rushSession: RushSession | undefined; + + public constructor(repoRoot: string, onDispose?: () => unknown) { + this.#onDispose = onDispose; + this.metadata = { + projectCount: 0, + projectNames: [], + repoRoot, + rushJsonFile: path.join(repoRoot, 'rush.json'), + rushVersion: '5.178.0' + }; + } + + public async [Symbol.asyncDispose](): Promise { + await this.#onDispose?.(); + } + + public reconcileInvalidationsAsync(): Promise { + return Promise.resolve(undefined); + } +} diff --git a/libraries/rush-daemon/src/test/WorkspaceEngineComponentFactory.test.ts b/libraries/rush-daemon/src/test/WorkspaceEngineComponentFactory.test.ts new file mode 100644 index 0000000000..43ad4959eb --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceEngineComponentFactory.test.ts @@ -0,0 +1,498 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IInputsSnapshot, + IOperationExecutionResult, + IOperationGraph, + IPhase, + Parallelism, + RushConfigurationProject +} from '@microsoft/rush-lib'; +import { + Operation, + OperationGraphHooks, + OperationStatus, + RushSession +} from '@microsoft/rush-lib'; + +import { + WorkspaceEngineComponentFactory +} from '../WorkspaceEngineComponentFactory'; +import type { + ICreateWorkspaceEngineComponentsOptions, + IMapWorkspaceInvalidationsOptions, + IWorkspaceEngineComponents, + IWorkspaceEngineShape +} from '../WorkspaceEngineComponentFactory'; +import { WorkspaceSession } from '../WorkspaceSession'; +import type { + IWorkspaceInvalidationWatcher, + IWorkspaceSessionComponents +} from '../WorkspaceSession'; +import { WorkspaceInvalidationTracker } from '../WorkspaceInvalidationTracker'; +import { TEST_RUSH_CONFIGURATION, TEST_REPO_ROOT } from './TestWorkspaceSession'; + +const PHASE_NAME: string = '_phase:test'; +const PLUGIN_NAME: string = '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: PHASE_NAME +}; + +interface ITestEngine { + readonly components: IWorkspaceEngineComponents; + readonly graph: TestOperationGraph; + readonly operations: ReadonlyArray; +} + +class TestOperationGraph implements IOperationGraph { + #parallelism: number = 1; + + public readonly abortController: AbortController = new AbortController(); + public readonly hooks: OperationGraphHooks = new OperationGraphHooks(); + public readonly resultByOperation: ReadonlyMap = new Map(); + public readonly status: OperationStatus = OperationStatus.Ready; + public readonly terminalDestinations: IOperationGraph['terminalDestinations'] = new Set(); + public allowOversubscription: boolean = true; + public debugMode: boolean = false; + public hasScheduledIteration: boolean = false; + public pauseNextIteration: boolean = false; + public quietMode: boolean = true; + public readonly operations: ReadonlySet; + + public constructor(operations: ReadonlySet) { + this.operations = operations; + } + + public get parallelism(): number { + return this.#parallelism; + } + + public set parallelism(value: Parallelism) { + this.#parallelism = typeof value === 'number' ? value : 1; + } + + public abortCurrentIterationAsync(): Promise { + return Promise.resolve(); + } + + public addTerminalDestination(): void {} + + public closeRunnersAsync(): Promise { + return Promise.resolve(); + } + + public executeScheduledIterationAsync(): Promise { + return Promise.resolve(false); + } + + public invalidateOperations(): void {} + + public removeTerminalDestination(): boolean { + return false; + } + + public scheduleIterationAsync(): Promise { + return Promise.resolve(false); + } + + public setEnabledStates(): boolean { + return false; + } +} + +function createInputsSnapshot(name: string): IInputsSnapshot { + const hashes: ReadonlyMap = new Map([[`${name}.ts`, name]]); + return { + getOperationOwnStateHash: () => name, + getTrackedFileHashesForOperation: () => hashes, + hasUncommittedChanges: true, + hashes, + rootDirectory: TEST_REPO_ROOT + }; +} + +function createTestEngine( + getInputsSnapshotAsync: () => Promise, + onDisposeAsync?: () => Promise +): ITestEngine { + const operations: Operation[] = Array.from( + TEST_RUSH_CONFIGURATION.projects, + (project: RushConfigurationProject) => + new Operation({ + logFilenameIdentifier: '_phase_test', + phase: TEST_PHASE, + project + }) + ); + const graph: TestOperationGraph = new TestOperationGraph(new Set(operations)); + const disposeEngineAsync = async (): Promise => { + const errors: unknown[] = []; + graph.abortController.abort(); + for (const cleanupAsync of [ + () => graph.abortCurrentIterationAsync(), + () => graph.closeRunnersAsync(), + onDisposeAsync + ]) { + if (!cleanupAsync) { + continue; + } + try { + await cleanupAsync(); + } catch (error) { + errors.push(error); + } + } + if (errors.length === 1) { + throw errors[0]; + } else if (errors.length > 1) { + throw new AggregateError(errors, 'Failed to dispose test engine components.'); + } + }; + const components: IWorkspaceEngineComponents = { + [Symbol.asyncDispose]: disposeEngineAsync, + getInputsSnapshotAsync, + inputsSnapshot: createInputsSnapshot('initial'), + operationGraph: graph, + rushSession: new RushSession({ + getIsDebugMode: () => false, + terminalProvider: { + eolCharacter: '\n', + supportsColor: false, + write: () => undefined + } + }) + }; + return { components, graph, operations }; +} + +function getReconcileAsync( + components: IWorkspaceSessionComponents +): NonNullable { + const reconcileAsync: IWorkspaceSessionComponents['reconcileInvalidationsAsync'] = + components.reconcileInvalidationsAsync; + if (!reconcileAsync) { + throw new Error('Expected workspace reconciliation to be configured.'); + } + return reconcileAsync; +} + +async function disposeComponentsAsync(components: IWorkspaceSessionComponents): Promise { + await components[Symbol.asyncDispose](); +} + +describe(WorkspaceEngineComponentFactory.name, () => { + it('initializes exactly once through WorkspaceSession and reconciles startup conservatively', async () => { + const nextSnapshot: IInputsSnapshot = createInputsSnapshot('next'); + const engine: ITestEngine = createTestEngine(() => Promise.resolve(nextSnapshot)); + const createEngineComponentsAsync: jest.Mock< + Promise, + [ICreateWorkspaceEngineComponentsOptions] + > = jest.fn(async (createOptions: ICreateWorkspaceEngineComponentsOptions) => { + void createOptions; + return engine.components; + }); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync, + mapInvalidationsToOperationsAsync: async () => [], + shape: { + phaseNames: [PHASE_NAME], + pluginNames: [] + } + }); + const watcher: IWorkspaceInvalidationWatcher = { + [Symbol.asyncDispose]: () => Promise.resolve(), + startAsync: () => Promise.resolve() + }; + const invalidateSpy: jest.SpyInstance = jest.spyOn(engine.graph, 'invalidateOperations'); + const session: WorkspaceSession = await WorkspaceSession.createAsync({ + createComponentsAsync: async (createOptions) => { + const engineComponents: IWorkspaceSessionComponents = + await factory.createAsync(createOptions); + return { + ...engineComponents, + projectWatcher: watcher, + [Symbol.asyncDispose]: async () => { + await watcher[Symbol.asyncDispose](); + await engineComponents[Symbol.asyncDispose](); + } + }; + }, + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.1' + }); + + const result = await session.reconcileInvalidationsAsync(); + + expect(createEngineComponentsAsync).toHaveBeenCalledTimes(1); + expect(session.operationGraph).toBe(engine.graph); + expect(session.engineShape).toEqual({ + phaseNames: [PHASE_NAME], + pluginNames: [] + }); + expect(result).toMatchObject({ + inputsSnapshot: nextSnapshot, + invalidatedOperationCount: engine.operations.length, + isFullInvalidation: true, + sequence: 1 + }); + expect(session.inputsSnapshot).toBe(nextSnapshot); + expect(invalidateSpy).toHaveBeenCalledWith(undefined, 'workspace-inputs-changed'); + await session[Symbol.asyncDispose](); + }); + + it('constructs an explicitly shaped all-project engine and maps retained paths', async () => { + const nextSnapshot: IInputsSnapshot = createInputsSnapshot('next'); + const engine: ITestEngine = createTestEngine(() => Promise.resolve(nextSnapshot)); + const targetOperation: Operation = engine.operations[0]; + const createEngineComponentsAsync: jest.Mock< + Promise, + [ICreateWorkspaceEngineComponentsOptions] + > = jest.fn(async (createOptions: ICreateWorkspaceEngineComponentsOptions) => { + void createOptions; + return engine.components; + }); + const mapInvalidationsToOperationsAsync: jest.Mock< + Promise>, + [IMapWorkspaceInvalidationsOptions] + > = jest.fn(async (mapOptions: IMapWorkspaceInvalidationsOptions) => { + void mapOptions; + return [targetOperation, targetOperation]; + }); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync, + mapInvalidationsToOperationsAsync, + shape: { + phaseNames: [PHASE_NAME], + pluginNames: [PLUGIN_NAME] + } + }); + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + invalidations.invalidate('libraries/a/src/index.ts'); + const components: IWorkspaceSessionComponents = await factory.createAsync({ + invalidations, + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + const invalidateSpy: jest.SpyInstance = jest.spyOn(engine.graph, 'invalidateOperations'); + + const result = await getReconcileAsync(components)(); + + const createOptions: ICreateWorkspaceEngineComponentsOptions = + createEngineComponentsAsync.mock.calls[0][0]; + expect(createOptions.projectSelection).toEqual(new Set(TEST_RUSH_CONFIGURATION.projects)); + expect(createOptions.phaseNames).toEqual([PHASE_NAME]); + expect(createOptions.pluginNames).toEqual([PLUGIN_NAME]); + expect(result).toMatchObject({ + inputsSnapshot: nextSnapshot, + invalidatedOperationCount: 1, + isFullInvalidation: false, + sequence: 1 + }); + expect(components.inputsSnapshot).toBe(nextSnapshot); + expect(invalidateSpy).toHaveBeenCalledWith(new Set([targetOperation]), 'workspace-inputs-changed'); + expect(invalidations.getSnapshot()).toMatchObject({ + changedPaths: [], + hasUnknownChanges: false + }); + await disposeComponentsAsync(components); + }); + + it('uses a full invalidation for unknown changes and for snapshot races', async () => { + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + invalidations.invalidate('libraries/a/src/index.ts'); + let snapshotCalls: number = 0; + const engine: ITestEngine = createTestEngine(() => { + snapshotCalls++; + if (snapshotCalls === 1) { + invalidations.invalidate('libraries/b/src/index.ts'); + } + return Promise.resolve(createInputsSnapshot(`next-${snapshotCalls}`)); + }); + const mapInvalidationsToOperationsAsync: jest.Mock = jest.fn(async () => [engine.operations[0]]); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => engine.components, + mapInvalidationsToOperationsAsync, + shape: { + phaseNames: [PHASE_NAME], + pluginNames: [PLUGIN_NAME] + } + }); + const components: IWorkspaceSessionComponents = await factory.createAsync({ + invalidations, + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + const invalidateSpy: jest.SpyInstance = jest.spyOn(engine.graph, 'invalidateOperations'); + + const firstResult = await getReconcileAsync(components)(); + const secondResult = await getReconcileAsync(components)(); + + expect(firstResult.isFullInvalidation).toBe(false); + expect(secondResult).toMatchObject({ + invalidatedOperationCount: engine.operations.length, + isFullInvalidation: true, + sequence: 2 + }); + expect(mapInvalidationsToOperationsAsync).toHaveBeenCalledTimes(1); + expect(invalidateSpy).toHaveBeenLastCalledWith(undefined, 'workspace-inputs-changed'); + await disposeComponentsAsync(components); + }); + + it('retains invalidations when a mapper returns an operation outside the graph', async () => { + const engine: ITestEngine = createTestEngine(() => + Promise.resolve(createInputsSnapshot('next')) + ); + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + invalidations.invalidate('libraries/a/src/index.ts'); + const outsider: Operation = new Operation({ + logFilenameIdentifier: '_phase_test', + phase: TEST_PHASE, + project: TEST_RUSH_CONFIGURATION.projects[0] + }); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => engine.components, + mapInvalidationsToOperationsAsync: async () => [outsider], + shape: { + phaseNames: [PHASE_NAME], + pluginNames: [PLUGIN_NAME] + } + }); + const components: IWorkspaceSessionComponents = await factory.createAsync({ + invalidations, + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + + await expect(getReconcileAsync(components)()).rejects.toThrow( + 'operation outside the graph' + ); + expect(components.inputsSnapshot).toBe(engine.components.inputsSnapshot); + expect(invalidations.getSnapshot().changedPaths).toEqual(['libraries/a/src/index.ts']); + await disposeComponentsAsync(components); + }); + + it('waits for reconciliation and aggregates deterministic graph cleanup failures', async () => { + const events: string[] = []; + let finishSnapshot: (() => void) | undefined; + const snapshotPromise: Promise = new Promise( + (resolve: (snapshot: IInputsSnapshot) => void) => { + finishSnapshot = () => { + events.push('snapshot'); + resolve(createInputsSnapshot('next')); + }; + } + ); + const engine: ITestEngine = createTestEngine( + () => snapshotPromise, + async () => { + events.push('components-dispose'); + throw new Error('component cleanup failed'); + } + ); + engine.graph.abortController.signal.addEventListener( + 'abort', + () => events.push('session-abort'), + { once: true } + ); + jest.spyOn(engine.graph, 'abortCurrentIterationAsync').mockImplementation(async () => { + events.push('iteration-abort'); + throw new Error('graph abort failed'); + }); + jest.spyOn(engine.graph, 'closeRunnersAsync').mockImplementation(async () => { + events.push('runners-close'); + throw new Error('runner cleanup failed'); + }); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => engine.components, + mapInvalidationsToOperationsAsync: async () => [engine.operations[0]], + shape: { + phaseNames: [PHASE_NAME], + pluginNames: [PLUGIN_NAME] + } + }); + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + invalidations.invalidate('libraries/a/src/index.ts'); + const components: IWorkspaceSessionComponents = await factory.createAsync({ + invalidations, + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + const reconciliationPromise: Promise = getReconcileAsync(components)(); + const disposalPromise: Promise = Promise.resolve(components[Symbol.asyncDispose]()); + finishSnapshot?.(); + + await reconciliationPromise; + await expect(disposalPromise).rejects.toThrow('Failed to dispose test engine components'); + expect(events).toEqual([ + 'snapshot', + 'session-abort', + 'iteration-abort', + 'runners-close', + 'components-dispose' + ]); + }); + + it('accepts an explicitly empty plugin shape', () => { + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => + createTestEngine(() => Promise.resolve(createInputsSnapshot('next'))).components, + mapInvalidationsToOperationsAsync: async () => [], + shape: { + phaseNames: [PHASE_NAME], + pluginNames: [] + } + }); + + expect(factory.shape.pluginNames).toEqual([]); + }); + + it('rejects a graph that does not represent every configured project', async () => { + const engine: ITestEngine = createTestEngine(() => + Promise.resolve(createInputsSnapshot('next')) + ); + const shape: IWorkspaceEngineShape = { + phaseNames: [PHASE_NAME], + pluginNames: [PLUGIN_NAME] + }; + const subsetGraph: TestOperationGraph = new TestOperationGraph(new Set([engine.operations[0]])); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => ({ + ...engine.components, + operationGraph: subsetGraph + }), + mapInvalidationsToOperationsAsync: async () => [], + shape + }); + + await expect( + factory.createAsync({ + invalidations: new WorkspaceInvalidationTracker(), + rushConfiguration: TEST_RUSH_CONFIGURATION + }) + ).rejects.toThrow('does not represent project'); + }); + + it('rejects a graph containing an undeclared plugin phase', async () => { + const engine: ITestEngine = createTestEngine(() => + Promise.resolve(createInputsSnapshot('next')) + ); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => engine.components, + mapInvalidationsToOperationsAsync: async () => [], + shape: { + phaseNames: ['_phase:other'], + pluginNames: [PLUGIN_NAME] + } + }); + + await expect( + factory.createAsync({ + invalidations: new WorkspaceInvalidationTracker(), + rushConfiguration: TEST_RUSH_CONFIGURATION + }) + ).rejects.toThrow('is not declared in the workspace engine shape'); + }); +}); diff --git a/libraries/rush-daemon/src/test/WorkspaceSession.test.ts b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts new file mode 100644 index 0000000000..cf51deccfe --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts @@ -0,0 +1,248 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IWorkspaceInvalidationWatcher, + IWorkspaceSessionComponents +} from '../WorkspaceSession'; +import { WorkspaceSession } from '../WorkspaceSession'; +import type { IWorkspaceInvalidationSnapshot } from '../WorkspaceInvalidationTracker'; +import { WorkspaceInvalidationTracker } from '../WorkspaceInvalidationTracker'; +import { TEST_REPO_ROOT } from './TestWorkspaceSession'; + +class TestInvalidationWatcher implements IWorkspaceInvalidationWatcher { + readonly #events: string[]; + #onInvalidation: ((changedPath?: string) => void) | undefined; + + public constructor(events: string[]) { + this.#events = events; + } + + public startAsync(onInvalidation: (changedPath?: string) => void): Promise { + this.#events.push('watcher-start'); + this.#onInvalidation = onInvalidation; + return Promise.resolve(); + } + + public invalidate(changedPath?: string): void { + if (!this.#onInvalidation) { + throw new Error('The test watcher is not running.'); + } + this.#onInvalidation(changedPath); + } + + public [Symbol.asyncDispose](): Promise { + this.#events.push('watcher-dispose'); + this.#onInvalidation = undefined; + return Promise.resolve(); + } +} + +describe(WorkspaceSession.name, () => { + it('loads stable metadata and retains headless invalidations until acknowledged', async () => { + const events: string[] = []; + const watcher: TestInvalidationWatcher = new TestInvalidationWatcher(events); + let componentFactoryCalls: number = 0; + const session: WorkspaceSession = await WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => { + componentFactoryCalls++; + return Promise.resolve({ + projectWatcher: watcher, + [Symbol.asyncDispose]: async () => { + await watcher[Symbol.asyncDispose](); + events.push('components-dispose'); + } + }); + } + }); + + expect(componentFactoryCalls).toBe(1); + expect(session.metadata).toMatchObject({ + projectCount: session.rushConfiguration.projects.length, + repoRoot: session.rushConfiguration.rushJsonFolder, + rushJsonFile: session.rushConfiguration.rushJsonFile, + rushVersion: '5.178.0' + }); + expect(session.metadata.projectNames).toEqual( + Array.from(session.rushConfiguration.projectsByName.keys()).sort() + ); + + const initialSnapshot: IWorkspaceInvalidationSnapshot = session.invalidations.getSnapshot(); + expect(initialSnapshot).toEqual({ + changedPaths: [], + hasUnknownChanges: true, + isWatcherHealthy: true, + sequence: 1 + }); + session.invalidations.acknowledgeThrough(initialSnapshot.sequence); + + watcher.invalidate('packages/a/src/index.ts'); + const firstSnapshot: IWorkspaceInvalidationSnapshot = session.invalidations.getSnapshot(); + watcher.invalidate('packages/a/src/index.ts'); + watcher.invalidate(); + session.invalidations.acknowledgeThrough(firstSnapshot.sequence); + + expect(session.invalidations.getSnapshot()).toEqual({ + changedPaths: ['packages/a/src/index.ts'], + hasUnknownChanges: true, + isWatcherHealthy: true, + sequence: 4 + }); + + await session[Symbol.asyncDispose](); + expect(events).toEqual(['watcher-start', 'watcher-dispose', 'components-dispose']); + await session[Symbol.asyncDispose](); + expect(events).toEqual(['watcher-start', 'watcher-dispose', 'components-dispose']); + }); + + it('does not allow a watcher error to be acknowledged as clean', async () => { + const watcher: TestInvalidationWatcher = new TestInvalidationWatcher([]); + const session: WorkspaceSession = await WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: watcher, + [Symbol.asyncDispose]: () => watcher[Symbol.asyncDispose]() + }) + }); + + session.invalidations.markWatcherUnhealthy(); + const snapshot: IWorkspaceInvalidationSnapshot = session.invalidations.getSnapshot(); + session.invalidations.acknowledgeThrough(snapshot.sequence); + + expect(session.invalidations.getSnapshot()).toMatchObject({ + hasUnknownChanges: true, + isWatcherHealthy: false + }); + await session[Symbol.asyncDispose](); + }); + + it('compacts excessive path changes into an unknown invalidation', () => { + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + for (let index: number = 0; index <= 10_000; index++) { + invalidations.invalidate(`packages/project-${index}/lib/output.js`); + } + + const overflowSnapshot: IWorkspaceInvalidationSnapshot = invalidations.getSnapshot(); + expect(overflowSnapshot).toEqual({ + changedPaths: [], + hasUnknownChanges: true, + isWatcherHealthy: true, + sequence: 10_001 + }); + + invalidations.invalidate('packages/later-change/src/index.ts'); + invalidations.acknowledgeThrough(overflowSnapshot.sequence); + expect(invalidations.getSnapshot()).toMatchObject({ + changedPaths: [], + hasUnknownChanges: true, + sequence: 10_002 + }); + invalidations.acknowledgeThrough(10_002); + expect(invalidations.getSnapshot().hasUnknownChanges).toBe(false); + }); + + it('disposes components when watcher startup fails', async () => { + const events: string[] = []; + const watcher: IWorkspaceInvalidationWatcher = { + startAsync: () => Promise.reject(new Error('watcher startup failed')), + [Symbol.asyncDispose]: () => { + events.push('watcher-dispose'); + return Promise.resolve(); + } + }; + + await expect( + WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: watcher, + [Symbol.asyncDispose]: async () => { + await watcher[Symbol.asyncDispose](); + events.push('components-dispose'); + } + }) + }) + ).rejects.toThrow('watcher startup failed'); + expect(events).toEqual(['watcher-dispose', 'components-dispose']); + }); + + it('preserves initialization and cleanup failures when watcher startup fails', async () => { + const watcher: IWorkspaceInvalidationWatcher = { + startAsync: () => Promise.reject(new Error('watcher startup failed')), + [Symbol.asyncDispose]: () => Promise.reject(new Error('watcher cleanup failed')) + }; + + let thrownError: unknown; + try { + await WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: watcher, + [Symbol.asyncDispose]: async () => { + let watcherError: unknown; + try { + await watcher[Symbol.asyncDispose](); + } catch (error) { + watcherError = error; + } + throw new AggregateError( + [watcherError, new Error('component cleanup failed')], + 'component bundle cleanup failed' + ); + } + }) + }); + } catch (error) { + thrownError = error; + } + + expect(thrownError).toBeInstanceOf(AggregateError); + const errors: unknown[] = (thrownError as AggregateError).errors; + expect(errors[0]).toEqual(new Error('watcher startup failed')); + expect(errors[1]).toBeInstanceOf(AggregateError); + expect((errors[1] as AggregateError).errors).toEqual([ + new Error('watcher cleanup failed'), + new Error('component cleanup failed') + ]); + }); + + it('rejects reconciliation as soon as disposal starts', async () => { + let finishWatcherDisposal: (() => void) | undefined; + const watcher: IWorkspaceInvalidationWatcher = { + startAsync: () => Promise.resolve(), + [Symbol.asyncDispose]: () => + new Promise((resolve: () => void) => { + finishWatcherDisposal = resolve; + }) + }; + const reconcileInvalidationsAsync: jest.Mock = jest.fn(() => + Promise.reject(new Error('Reconciliation must not start.')) + ); + const session: WorkspaceSession = await WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: watcher, + reconcileInvalidationsAsync, + [Symbol.asyncDispose]: () => watcher[Symbol.asyncDispose]() + }) + }); + + const disposalPromise: Promise = session[Symbol.asyncDispose](); + await expect(session.reconcileInvalidationsAsync()).rejects.toThrow( + 'workspace session is being disposed' + ); + expect(reconcileInvalidationsAsync).not.toHaveBeenCalled(); + finishWatcherDisposal?.(); + await disposalPromise; + }); +}); diff --git a/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts b/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts new file mode 100644 index 0000000000..1d8a364dc8 --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts @@ -0,0 +1,47 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type * as fs from 'node:fs'; +import { EventEmitter } from 'node:events'; + +import { WorkspaceSessionFileWatcher } from '../WorkspaceSessionFileWatcher'; +import { TEST_RUSH_CONFIGURATION } from './TestWorkspaceSession'; + +class TestFsWatcher extends EventEmitter { + public close(): void { + this.emit('close'); + } + + public ref(): this { + return this; + } + + public unref(): this { + return this; + } +} + +describe(WorkspaceSessionFileWatcher.name, () => { + it('watches every configured subspace config folder', async () => { + const watchedPaths: string[] = []; + const watcher: WorkspaceSessionFileWatcher = new WorkspaceSessionFileWatcher({ + rushConfiguration: TEST_RUSH_CONFIGURATION, + watchFactory: (folderPath: string) => { + watchedPaths.push(folderPath); + return new TestFsWatcher() as fs.FSWatcher; + } + }); + + await watcher.startAsync(() => {}); + + const subspaceConfigFolders: string[] = TEST_RUSH_CONFIGURATION.subspaces.map((subspace) => + subspace.getSubspaceConfigFolderPath() + ); + expect(subspaceConfigFolders.length).toBeGreaterThan(0); + expect(watchedPaths).toEqual( + expect.arrayContaining(subspaceConfigFolders) + ); + + await watcher[Symbol.asyncDispose](); + }); +}); diff --git a/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts new file mode 100644 index 0000000000..a5328edaac --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts @@ -0,0 +1,148 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IWorkspaceSession, IWorkspaceSessionOptions } from '../WorkspaceSession'; +import { WorkspaceSessionProvider } from '../WorkspaceSessionProvider'; +import { TestWorkspaceSession } from './TestWorkspaceSession'; + +const OPTIONS: IWorkspaceSessionOptions = { + repoRoot: 'repo', + rushVersion: '5.178.0' +}; + +describe(WorkspaceSessionProvider.name, () => { + it('shares concurrent initialization and reuses the result', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot); + let resolveFactory: ((value: IWorkspaceSession) => void) | undefined; + let factoryCalls: number = 0; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider(() => { + factoryCalls++; + return new Promise((resolve) => { + resolveFactory = resolve; + }); + }, OPTIONS); + + const first: Promise = provider.getSessionAsync(); + const second: Promise = provider.getSessionAsync(); + expect(first).toBe(second); + await Promise.resolve(); + expect(factoryCalls).toBe(1); + + resolveFactory?.(session); + await expect(first).resolves.toBe(session); + await expect(provider.getSessionAsync()).resolves.toBe(session); + expect(factoryCalls).toBe(1); + await provider[Symbol.asyncDispose](); + }); + + it('clears a failed initialization so a later attempt can retry', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot); + let factoryCalls: number = 0; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider(() => { + factoryCalls++; + return factoryCalls === 1 + ? Promise.reject(new Error('initialization failed')) + : Promise.resolve(session); + }, OPTIONS); + + await expect(provider.getSessionAsync()).rejects.toThrow('initialization failed'); + await expect(provider.getSessionAsync()).resolves.toBe(session); + expect(factoryCalls).toBe(2); + await provider[Symbol.asyncDispose](); + }); + + it('clears a synchronously thrown initialization so a later attempt can retry', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot); + let factoryCalls: number = 0; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider(() => { + factoryCalls++; + if (factoryCalls === 1) { + throw new Error('synchronous initialization failed'); + } + return Promise.resolve(session); + }, OPTIONS); + + await expect(provider.getSessionAsync()).rejects.toThrow('synchronous initialization failed'); + await expect(provider.getSessionAsync()).resolves.toBe(session); + expect(factoryCalls).toBe(2); + await provider[Symbol.asyncDispose](); + }); + + it('disposes a session that finishes initializing during shutdown', async () => { + const disposalEvents: string[] = []; + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot, () => + disposalEvents.push('session') + ); + let resolveFactory: ((value: IWorkspaceSession) => void) | undefined; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider( + () => + new Promise((resolve) => { + resolveFactory = resolve; + }), + OPTIONS + ); + + const initialization: Promise = provider.getSessionAsync(); + const disposal: Promise = provider[Symbol.asyncDispose](); + await Promise.resolve(); + resolveFactory?.(session); + + await expect(initialization).rejects.toThrow('disposed during initialization'); + await disposal; + expect(disposalEvents).toEqual(['session']); + }); + + it('surfaces a disposal failure from a session that finishes initializing during shutdown', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot, () => + Promise.reject(new Error('session cleanup failed')) + ); + let resolveFactory: ((value: IWorkspaceSession) => void) | undefined; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider( + () => + new Promise((resolve) => { + resolveFactory = resolve; + }), + OPTIONS + ); + + const initialization: Promise = provider.getSessionAsync(); + const disposal: Promise = provider[Symbol.asyncDispose](); + const initializationExpectation: Promise = expect(initialization).rejects.toThrow( + 'session cleanup failed' + ); + const disposalExpectation: Promise = expect(disposal).rejects.toThrow( + 'session cleanup failed' + ); + await Promise.resolve(); + resolveFactory?.(session); + + await Promise.all([initializationExpectation, disposalExpectation]); + }); + + it('surfaces a synchronous disposal failure during initialization shutdown', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot, () => { + throw new Error('synchronous session cleanup failed'); + }); + let resolveFactory: ((value: IWorkspaceSession) => void) | undefined; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider( + () => + new Promise((resolve) => { + resolveFactory = resolve; + }), + OPTIONS + ); + + const initialization: Promise = provider.getSessionAsync(); + const disposal: Promise = provider[Symbol.asyncDispose](); + const initializationExpectation: Promise = expect(initialization).rejects.toThrow( + 'synchronous session cleanup failed' + ); + const disposalExpectation: Promise = expect(disposal).rejects.toThrow( + 'synchronous session cleanup failed' + ); + await Promise.resolve(); + resolveFactory?.(session); + + await Promise.all([initializationExpectation, disposalExpectation]); + }); +});