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..0103b49ccb --- /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, fail-closed graph recreation boundaries, and a deterministic engine shutdown contract.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index aa8bb2d015..9277421b0e 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -6,15 +6,36 @@ /// +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 IClassifyWorkspaceInvalidationsOptions { + // (undocumented) + readonly changedPaths: ReadonlyArray; + // (undocumented) + readonly rushConfiguration: RushConfiguration; +} + +// @beta +export interface ICreateWorkspaceEngineComponentsOptions extends IWorkspaceEngineShape { + readonly projectSelection: ReadonlySet; + // (undocumented) + readonly rushConfiguration: RushConfiguration; +} + // @beta export interface ICreateWorkspaceSessionComponentsOptions { // (undocumented) @@ -25,6 +46,18 @@ export interface ICreateWorkspaceSessionComponentsOptions { 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 { // (undocumented) @@ -58,6 +91,54 @@ export interface IRushDaemonServeOptions extends IRushDaemonHostOptions { readonly shutdownSignal?: AbortSignal; } +// @beta +export type IsWorkspaceEngineRecreationRequiredAsync = (options: IClassifyWorkspaceInvalidationsOptions) => Promise; + +// @beta +export interface IWorkspaceEngineComponentFactoryOptions { + // (undocumented) + readonly createEngineComponentsAsync: CreateWorkspaceEngineComponentsAsync; + // (undocumented) + readonly isEngineRecreationRequiredAsync?: IsWorkspaceEngineRecreationRequiredAsync; + // (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; @@ -74,6 +155,8 @@ export interface IWorkspaceInvalidationWatcher extends AsyncDisposable { // @beta export interface IWorkspaceSession extends AsyncDisposable { + // (undocumented) + readonly engineShape: IWorkspaceEngineShape | undefined; // (undocumented) readonly inputsSnapshot: IInputsSnapshot | undefined; // (undocumented) @@ -83,6 +166,8 @@ export interface IWorkspaceSession extends AsyncDisposable { // (undocumented) readonly operationGraph: IOperationGraph | undefined; // (undocumented) + reconcileInvalidationsAsync(): Promise; + // (undocumented) readonly rushConfiguration: RushConfiguration; // (undocumented) readonly rushSession: RushSession | undefined; @@ -90,12 +175,16 @@ export interface IWorkspaceSession extends AsyncDisposable { // @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; } @@ -125,6 +214,9 @@ export interface IWorkspaceSessionOptions { readonly rushVersion: string; } +// @beta +export type MapWorkspaceInvalidationsToOperationsAsync = (options: IMapWorkspaceInvalidationsOptions) => Promise>; + // @public export enum RequestExclusivityClass { // (undocumented) @@ -171,11 +263,29 @@ 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 WorkspaceEngineRecreationRequiredError extends Error { + constructor(); +} + // @beta export class WorkspaceInvalidationTracker { acknowledgeThrough(sequence: number): void; getSnapshot(): IWorkspaceInvalidationSnapshot; + // @internal (undocumented) + get hasUnattributedUnknownChanges(): boolean; invalidate(changedPath?: string): void; + // @internal (undocumented) + invalidateForInitialization(): void; markWatcherUnhealthy(): void; } @@ -184,13 +294,16 @@ export class WorkspaceSession implements IWorkspaceSession { [Symbol.asyncDispose](): Promise; static createAsync(options: IWorkspaceSessionOptions): Promise; // (undocumented) - readonly inputsSnapshot: IInputsSnapshot | undefined; + 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) diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index a2055f0727..f8e6cab2db 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -12,7 +12,18 @@ The host loads `RushConfiguration` once before signaling readiness and keeps a h 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. Reusable operation graph, plugin, and input snapshot state can be supplied -through the session component factory; the default session does not construct those command-specific -resources while the reusable runner lifetime tracked by -[rushstack#5895](https://github.com/microsoft/rushstack/issues/5895) remains incomplete. +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. +After the initial conservative startup reconciliation, changes to Rush configuration, project package manifests, +or integration-classified plugin graph inputs fail closed with `WorkspaceEngineRecreationRequiredError` before +the input baseline advances or the invalidation is acknowledged. The startup watcher-registration boundary has +no paths to classify and therefore remains a full invalidation. The routing layer must replace the complete +workspace session rather than run a stale graph. +The default daemon executable does not construct or route this graph while the command-independent plugin shape and per-iteration runner +lifetime tracked by [rushstack#5895](https://github.com/microsoft/rushstack/issues/5895) remain incomplete. diff --git a/libraries/rush-daemon/src/WorkspaceEngineComponentFactory.ts b/libraries/rush-daemon/src/WorkspaceEngineComponentFactory.ts new file mode 100644 index 0000000000..61d97935cd --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceEngineComponentFactory.ts @@ -0,0 +1,534 @@ +// 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 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: 'workspace-inputs-changed' = '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; +} + +/** + * Context for identifying changes that require a new workspace engine. + * + * @beta + */ +export interface IClassifyWorkspaceInvalidationsOptions { + readonly changedPaths: ReadonlyArray; + readonly rushConfiguration: RushConfiguration; +} + +/** + * Identifies integration-specific graph inputs that cannot be reconciled against an existing graph. + * + * @remarks + * Rush configuration files and project package manifests are classified automatically. Use this callback + * for plugin-specific graph inputs outside those locations. + * + * @beta + */ +export type IsWorkspaceEngineRecreationRequiredAsync = ( + options: IClassifyWorkspaceInvalidationsOptions +) => Promise; + +/** + * 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; +} + +/** + * Indicates that retained changes require the owning workspace session to be recreated. + * + * @remarks + * The invalidations remain unacknowledged. Callers must not execute the existing operation graph after + * receiving this error. + * + * @beta + */ +export class WorkspaceEngineRecreationRequiredError extends Error { + public constructor() { + super('Workspace changes require the reusable engine and session to be recreated.'); + this.name = 'WorkspaceEngineRecreationRequiredError'; + } +} + +/** + * Options for {@link WorkspaceEngineComponentFactory}. + * + * @beta + */ +export interface IWorkspaceEngineComponentFactoryOptions { + readonly createEngineComponentsAsync: CreateWorkspaceEngineComponentsAsync; + readonly isEngineRecreationRequiredAsync?: IsWorkspaceEngineRecreationRequiredAsync; + readonly mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; + readonly shape: IWorkspaceEngineShape; +} + +interface IWorkspaceEngineLifecycleOptions { + readonly components: IWorkspaceEngineComponents; + readonly graphDefiningPaths: IGraphDefiningPaths; + readonly invalidations: WorkspaceInvalidationTracker; + readonly isEngineRecreationRequiredAsync: IsWorkspaceEngineRecreationRequiredAsync | undefined; + readonly mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; + readonly rushConfiguration: RushConfiguration; +} + +interface IGraphDefiningPaths { + readonly filePaths: ReadonlySet; + readonly folderPaths: ReadonlyArray; +} + +class WorkspaceEngineLifecycle { + readonly #components: IWorkspaceEngineComponents; + readonly #graphDefiningPaths: IGraphDefiningPaths; + readonly #invalidations: WorkspaceInvalidationTracker; + readonly #isEngineRecreationRequiredAsync: IsWorkspaceEngineRecreationRequiredAsync | undefined; + readonly #mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; + readonly #rushConfiguration: RushConfiguration; + #currentInputsSnapshot: IInputsSnapshot; + #disposePromise: Promise | undefined; + #isDisposing: boolean = false; + #reconciliationTail: Promise = Promise.resolve(); + #requiresFullInvalidation: boolean = false; + + public constructor(options: IWorkspaceEngineLifecycleOptions) { + this.#components = options.components; + this.#graphDefiningPaths = options.graphDefiningPaths; + this.#currentInputsSnapshot = options.components.inputsSnapshot; + this.#invalidations = options.invalidations; + this.#isEngineRecreationRequiredAsync = options.isEngineRecreationRequiredAsync; + this.#mapInvalidationsToOperationsAsync = options.mapInvalidationsToOperationsAsync; + this.#rushConfiguration = options.rushConfiguration; + } + + 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(); + if (await this.#requiresEngineRecreationAsync(invalidationSnapshot)) { + throw new WorkspaceEngineRecreationRequiredError(); + } + 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](); + } + + async #requiresEngineRecreationAsync( + invalidationSnapshot: IWorkspaceInvalidationSnapshot + ): Promise { + if ( + invalidationSnapshot.changedPaths.length === 0 && + !invalidationSnapshot.hasUnknownChanges && + invalidationSnapshot.isWatcherHealthy + ) { + return false; + } + if ( + this.#invalidations.hasUnattributedUnknownChanges || + !invalidationSnapshot.isWatcherHealthy + ) { + return true; + } + + if ( + invalidationSnapshot.changedPaths.some((changedPath: string) => + isBuiltInGraphDefiningPath( + changedPath, + this.#rushConfiguration.rushJsonFolder, + this.#graphDefiningPaths + ) + ) + ) { + return true; + } + + return ( + (await this.#isEngineRecreationRequiredAsync?.({ + changedPaths: invalidationSnapshot.changedPaths, + rushConfiguration: this.#rushConfiguration + })) ?? false + ); + } +} + +/** + * 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 #isEngineRecreationRequiredAsync: IsWorkspaceEngineRecreationRequiredAsync | undefined; + readonly #mapInvalidationsToOperationsAsync: MapWorkspaceInvalidationsToOperationsAsync; + + public readonly createAsync: CreateWorkspaceSessionComponentsAsync; + public readonly shape: IWorkspaceEngineShape; + + public constructor(options: IWorkspaceEngineComponentFactoryOptions) { + this.#createEngineComponentsAsync = options.createEngineComponentsAsync; + this.#isEngineRecreationRequiredAsync = options.isEngineRecreationRequiredAsync; + 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 graphDefiningPaths: IGraphDefiningPaths = createGraphDefiningPaths( + options.rushConfiguration + ); + 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, + graphDefiningPaths, + invalidations: options.invalidations, + isEngineRecreationRequiredAsync: this.#isEngineRecreationRequiredAsync, + mapInvalidationsToOperationsAsync: this.#mapInvalidationsToOperationsAsync, + rushConfiguration: options.rushConfiguration + }); + 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 createGraphDefiningPaths(rushConfiguration: RushConfiguration): IGraphDefiningPaths { + const filePaths: Set = new Set([path.resolve(rushConfiguration.rushJsonFile)]); + for (const project of rushConfiguration.projects) { + filePaths.add(path.join(project.projectFolder, 'package.json')); + filePaths.add(path.join(project.projectFolder, 'config', 'rush-project.json')); + } + return { + filePaths, + folderPaths: [ + path.resolve(rushConfiguration.commonRushConfigFolder), + ...Array.from(rushConfiguration.subspaces, (subspace) => + path.resolve(subspace.getSubspaceConfigFolderPath()) + ) + ] + }; +} + +function isBuiltInGraphDefiningPath( + changedPath: string, + rushJsonFolder: string, + graphDefiningPaths: IGraphDefiningPaths +): boolean { + const absoluteChangedPath: string = path.resolve(rushJsonFolder, changedPath); + if (graphDefiningPaths.filePaths.has(absoluteChangedPath)) { + return true; + } + for (const folderPath of graphDefiningPaths.folderPaths) { + if (isPathInside(absoluteChangedPath, folderPath)) { + return true; + } + } + return false; +} + +function isPathInside(candidatePath: string, folderPath: string): boolean { + const relativePath: string = path.relative(path.resolve(folderPath), candidatePath); + return ( + relativePath === '' || + (!path.isAbsolute(relativePath) && + relativePath !== '..' && + !relativePath.startsWith(`..${path.sep}`)) + ); +} + +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 index ee67d5afa1..d12aedc38f 100644 --- a/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts +++ b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts @@ -9,7 +9,7 @@ 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. */ + /** True at initialization, or when the watcher reported a change without a path or encountered an error. */ readonly hasUnknownChanges: boolean; /** False after a watcher error makes subsequent change detection unreliable. */ readonly isWatcherHealthy: boolean; @@ -26,6 +26,7 @@ const MAX_TRACKED_CHANGED_PATHS: number = 10_000; */ export class WorkspaceInvalidationTracker { readonly #sequenceByPath: Map = new Map(); + #initializationSequence: number | undefined; #latestSequence: number = 0; #unknownChangeSequence: number | undefined; #watcherHealthy: boolean = true; @@ -50,6 +51,16 @@ export class WorkspaceInvalidationTracker { this.#sequenceByPath.set(changedPath, sequence); } + /** @internal */ + public invalidateForInitialization(): void { + this.#initializationSequence = ++this.#latestSequence; + } + + /** @internal */ + public get hasUnattributedUnknownChanges(): boolean { + return this.#unknownChangeSequence !== undefined; + } + /** * Permanently marks the current watcher as unhealthy. * @@ -66,7 +77,8 @@ export class WorkspaceInvalidationTracker { public getSnapshot(): IWorkspaceInvalidationSnapshot { return { changedPaths: Array.from(this.#sequenceByPath.keys()).sort(), - hasUnknownChanges: this.#unknownChangeSequence !== undefined, + hasUnknownChanges: + this.#initializationSequence !== undefined || this.#unknownChangeSequence !== undefined, isWatcherHealthy: this.#watcherHealthy, sequence: this.#latestSequence }; @@ -94,5 +106,8 @@ export class WorkspaceInvalidationTracker { ) { this.#unknownChangeSequence = undefined; } + if (this.#initializationSequence !== undefined && this.#initializationSequence <= sequence) { + this.#initializationSequence = undefined; + } } } diff --git a/libraries/rush-daemon/src/WorkspaceSession.ts b/libraries/rush-daemon/src/WorkspaceSession.ts index 8abe498e1f..12cd66c20f 100644 --- a/libraries/rush-daemon/src/WorkspaceSession.ts +++ b/libraries/rush-daemon/src/WorkspaceSession.ts @@ -12,6 +12,10 @@ import type { import { WorkspaceInvalidationTracker } from './WorkspaceInvalidationTracker'; import { WorkspaceSessionFileWatcher } from './WorkspaceSessionFileWatcher'; +import type { + IWorkspaceEngineShape, + IWorkspaceInvalidationReconciliation +} from './WorkspaceEngineComponentFactory'; /** * Stable identity loaded once for a warm workspace session. @@ -45,6 +49,7 @@ export interface IWorkspaceInvalidationWatcher extends AsyncDisposable { * @beta */ export interface IWorkspaceSessionComponents extends AsyncDisposable { + readonly engineShape?: IWorkspaceEngineShape; readonly inputsSnapshot?: IInputsSnapshot; readonly operationGraph?: IOperationGraph; /** @@ -55,6 +60,7 @@ export interface IWorkspaceSessionComponents extends AsyncDisposable { * `WorkspaceSession` directly disposes only the default watcher that it creates itself. */ readonly projectWatcher?: IWorkspaceInvalidationWatcher; + readonly reconcileInvalidationsAsync?: () => Promise; readonly rushSession?: RushSession; } @@ -96,12 +102,14 @@ export interface IWorkspaceSessionOptions { * @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; } /** @@ -124,8 +132,9 @@ export class WorkspaceSession implements IWorkspaceSession { readonly #components: IWorkspaceSessionComponents; readonly #sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined; #disposePromise: Promise | undefined; + #inputsSnapshot: IInputsSnapshot | undefined; + #isDisposing: boolean = false; - public readonly inputsSnapshot: IInputsSnapshot | undefined; public readonly invalidations: WorkspaceInvalidationTracker; public readonly metadata: IWorkspaceSessionMetadata; public readonly operationGraph: IOperationGraph | undefined; @@ -144,11 +153,19 @@ export class WorkspaceSession implements IWorkspaceSession { this.invalidations = invalidations; this.#components = components; this.#sessionOwnedProjectWatcher = sessionOwnedProjectWatcher; - this.inputsSnapshot = components.inputsSnapshot; + 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( @@ -194,7 +211,7 @@ export class WorkspaceSession implements IWorkspaceSession { invalidations.invalidate(changedPath) ); // Changes before the watcher registered its callbacks cannot be observed path-by-path. - invalidations.invalidate(); + invalidations.invalidateForInitialization(); return session; } catch (error) { const cleanupErrors: unknown[] = []; @@ -220,10 +237,25 @@ export class WorkspaceSession implements IWorkspaceSession { /** 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 { diff --git a/libraries/rush-daemon/src/index.ts b/libraries/rush-daemon/src/index.ts index d0d438199b..2de9d94142 100644 --- a/libraries/rush-daemon/src/index.ts +++ b/libraries/rush-daemon/src/index.ts @@ -13,6 +13,20 @@ export { } from './RequestScheduler'; export { RushDaemonHost, type IRushDaemonHostOptions } from './RushDaemonHost'; export { serveRushDaemonAsync, type IRushDaemonServeOptions } from './serveRushDaemon'; +export { + WorkspaceEngineComponentFactory, + WorkspaceEngineRecreationRequiredError, + type CreateWorkspaceEngineComponentsAsync, + type IClassifyWorkspaceInvalidationsOptions, + type ICreateWorkspaceEngineComponentsOptions, + type IMapWorkspaceInvalidationsOptions, + type IWorkspaceEngineComponentFactoryOptions, + type IWorkspaceEngineComponents, + type IWorkspaceEngineShape, + type IWorkspaceInvalidationReconciliation, + type IsWorkspaceEngineRecreationRequiredAsync, + type MapWorkspaceInvalidationsToOperationsAsync +} from './WorkspaceEngineComponentFactory'; export { WorkspaceSession, type CreateWorkspaceSessionComponentsAsync, diff --git a/libraries/rush-daemon/src/test/TestWorkspaceSession.ts b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts index fb609e0c73..4f917d98ff 100644 --- a/libraries/rush-daemon/src/test/TestWorkspaceSession.ts +++ b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts @@ -10,6 +10,10 @@ import type { RushSession } from '@microsoft/rush-lib'; +import type { + IWorkspaceEngineShape, + IWorkspaceInvalidationReconciliation +} from '../WorkspaceEngineComponentFactory'; import type { IWorkspaceSession, IWorkspaceSessionMetadata @@ -24,6 +28,7 @@ export const TEST_RUSH_CONFIGURATION: RushConfiguration = RushConfiguration.load 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; @@ -45,4 +50,8 @@ export class TestWorkspaceSession implements IWorkspaceSession { 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..0ed08d6a33 --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceEngineComponentFactory.test.ts @@ -0,0 +1,756 @@ +// 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 type { + IInputsSnapshot, + IOperationExecutionResult, + IOperationGraph, + IPhase, + Parallelism, + RushConfigurationProject +} from '@microsoft/rush-lib'; +import { + Operation, + OperationGraphHooks, + OperationStatus, + RushSession +} from '@microsoft/rush-lib'; + +import { + WorkspaceEngineComponentFactory, + WorkspaceEngineRecreationRequiredError +} from '../WorkspaceEngineComponentFactory'; +import type { + IClassifyWorkspaceInvalidationsOptions, + 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( + projects: Iterable, + getInputsSnapshotAsync: () => Promise, + onDisposeAsync?: () => Promise +): ITestEngine { + const operations: Operation[] = Array.from( + 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'); + let initializedEngine: ITestEngine | undefined; + let initializedRushConfiguration: + | ICreateWorkspaceEngineComponentsOptions['rushConfiguration'] + | undefined; + const createEngineComponentsAsync: jest.Mock< + Promise, + [ICreateWorkspaceEngineComponentsOptions] + > = jest.fn(async (createOptions: ICreateWorkspaceEngineComponentsOptions) => { + initializedRushConfiguration = createOptions.rushConfiguration; + initializedEngine = createTestEngine(createOptions.rushConfiguration.projects, () => + Promise.resolve(nextSnapshot) + ); + return initializedEngine.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 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 engine: ITestEngine | undefined = initializedEngine; + if (!engine) { + throw new Error('Expected the workspace engine to be initialized.'); + } + const invalidateSpy: jest.SpyInstance = jest.spyOn(engine.graph, 'invalidateOperations'); + + const result = await session.reconcileInvalidationsAsync(); + + expect(createEngineComponentsAsync).toHaveBeenCalledTimes(1); + expect(initializedRushConfiguration).toBe(session.rushConfiguration); + expect(session.operationGraph).toBe(engine.graph); + expect( + engine.operations.every((operation: Operation) => + session.rushConfiguration.projects.includes(operation.associatedProject) + ) + ).toBe(true); + 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(TEST_RUSH_CONFIGURATION.projects, () => + 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('serializes concurrent reconciliation against the latest inputs snapshot', async () => { + const initialSnapshot: IInputsSnapshot = createInputsSnapshot('initial'); + const firstSnapshot: IInputsSnapshot = createInputsSnapshot('first'); + const secondSnapshot: IInputsSnapshot = createInputsSnapshot('second'); + let startFirstSnapshot: (() => void) | undefined; + const firstSnapshotStarted: Promise = new Promise((resolve: () => void) => { + startFirstSnapshot = resolve; + }); + let finishFirstSnapshot: (() => void) | undefined; + const blockedFirstSnapshot: Promise = new Promise( + (resolve: (snapshot: IInputsSnapshot) => void) => { + finishFirstSnapshot = () => resolve(firstSnapshot); + } + ); + let snapshotCalls: number = 0; + const engine: ITestEngine = createTestEngine(TEST_RUSH_CONFIGURATION.projects, () => { + snapshotCalls++; + if (snapshotCalls === 1) { + startFirstSnapshot?.(); + return blockedFirstSnapshot; + } + return Promise.resolve(secondSnapshot); + }); + const mapInvalidationsToOperationsAsync: jest.Mock< + Promise>, + [IMapWorkspaceInvalidationsOptions] + > = jest.fn(async (mapOptions: IMapWorkspaceInvalidationsOptions) => { + void mapOptions; + return [engine.operations[0]]; + }); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => ({ + ...engine.components, + inputsSnapshot: initialSnapshot + }), + 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 firstReconciliation: Promise = getReconcileAsync(components)(); + await firstSnapshotStarted; + const queueSecondInvalidation: Promise = firstReconciliation.then(() => { + invalidations.invalidate('libraries/b/src/index.ts'); + }); + const secondReconciliation: Promise = getReconcileAsync(components)(); + + expect(snapshotCalls).toBe(1); + expect(mapInvalidationsToOperationsAsync).not.toHaveBeenCalled(); + finishFirstSnapshot?.(); + await Promise.all([firstReconciliation, queueSecondInvalidation, secondReconciliation]); + + expect(mapInvalidationsToOperationsAsync).toHaveBeenCalledTimes(2); + expect(mapInvalidationsToOperationsAsync.mock.calls[0][0]).toMatchObject({ + currentInputsSnapshot: initialSnapshot, + nextInputsSnapshot: firstSnapshot + }); + expect(mapInvalidationsToOperationsAsync.mock.calls[1][0]).toMatchObject({ + currentInputsSnapshot: firstSnapshot, + nextInputsSnapshot: secondSnapshot + }); + expect(components.inputsSnapshot).toBe(secondSnapshot); + 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(TEST_RUSH_CONFIGURATION.projects, () => { + 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 graph-defining invalidations and requires session recreation', async () => { + const getInputsSnapshotAsync: jest.Mock = jest.fn(async () => createInputsSnapshot('next')); + const engine: ITestEngine = createTestEngine( + TEST_RUSH_CONFIGURATION.projects, + getInputsSnapshotAsync + ); + 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 invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + const changedPath: string = path.join( + TEST_RUSH_CONFIGURATION.projects[0].projectFolder, + 'package.json' + ); + invalidations.invalidate(changedPath); + const components: IWorkspaceSessionComponents = await factory.createAsync({ + invalidations, + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + const invalidateSpy: jest.SpyInstance = jest.spyOn(engine.graph, 'invalidateOperations'); + + await expect(getReconcileAsync(components)()).rejects.toBeInstanceOf( + WorkspaceEngineRecreationRequiredError + ); + expect(getInputsSnapshotAsync).not.toHaveBeenCalled(); + expect(mapInvalidationsToOperationsAsync).not.toHaveBeenCalled(); + expect(invalidateSpy).not.toHaveBeenCalled(); + expect(components.inputsSnapshot).toBe(engine.components.inputsSnapshot); + expect(invalidations.getSnapshot().changedPaths).toEqual([changedPath]); + await disposeComponentsAsync(components); + }); + + it('supports integration-specific graph input classification', async () => { + const engine: ITestEngine = createTestEngine(TEST_RUSH_CONFIGURATION.projects, () => + Promise.resolve(createInputsSnapshot('next')) + ); + const changedPath: string = path.join(TEST_REPO_ROOT, 'config', 'test-plugin.json'); + const isEngineRecreationRequiredAsync: jest.Mock< + Promise, + [IClassifyWorkspaceInvalidationsOptions] + > = jest.fn(async (options: IClassifyWorkspaceInvalidationsOptions) => { + void options; + return true; + }); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => engine.components, + isEngineRecreationRequiredAsync, + mapInvalidationsToOperationsAsync: async () => [], + shape: { + phaseNames: [PHASE_NAME], + pluginNames: [PLUGIN_NAME] + } + }); + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + invalidations.invalidate(changedPath); + const components: IWorkspaceSessionComponents = await factory.createAsync({ + invalidations, + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + + await expect(getReconcileAsync(components)()).rejects.toBeInstanceOf( + WorkspaceEngineRecreationRequiredError + ); + expect(isEngineRecreationRequiredAsync).toHaveBeenCalledWith({ + changedPaths: [changedPath], + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + expect(invalidations.getSnapshot().changedPaths).toEqual([changedPath]); + await disposeComponentsAsync(components); + }); + + it('requires recreation for unknown changes after the startup baseline', async () => { + const getInputsSnapshotAsync: jest.Mock = jest.fn(async () => createInputsSnapshot('next')); + const engine: ITestEngine = createTestEngine( + TEST_RUSH_CONFIGURATION.projects, + getInputsSnapshotAsync + ); + const factory: WorkspaceEngineComponentFactory = new WorkspaceEngineComponentFactory({ + createEngineComponentsAsync: async () => engine.components, + mapInvalidationsToOperationsAsync: async () => [], + shape: { + phaseNames: [PHASE_NAME], + pluginNames: [PLUGIN_NAME] + } + }); + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + invalidations.invalidateForInitialization(); + const components: IWorkspaceSessionComponents = await factory.createAsync({ + invalidations, + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + + await expect(getReconcileAsync(components)()).resolves.toMatchObject({ + isFullInvalidation: true, + sequence: 1 + }); + invalidations.invalidate(); + await expect(getReconcileAsync(components)()).rejects.toBeInstanceOf( + WorkspaceEngineRecreationRequiredError + ); + expect(getInputsSnapshotAsync).toHaveBeenCalledTimes(1); + expect(invalidations.getSnapshot()).toMatchObject({ + hasUnknownChanges: true, + sequence: 2 + }); + await disposeComponentsAsync(components); + }); + + it('classifies known changes that arrive before the startup baseline', async () => { + const engine: ITestEngine = createTestEngine(TEST_RUSH_CONFIGURATION.projects, () => + Promise.resolve(createInputsSnapshot('next')) + ); + 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 invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + invalidations.invalidateForInitialization(); + const changedPath: string = path.join( + TEST_RUSH_CONFIGURATION.projects[0].projectFolder, + 'src', + 'index.ts' + ); + invalidations.invalidate(changedPath); + const components: IWorkspaceSessionComponents = await factory.createAsync({ + invalidations, + rushConfiguration: TEST_RUSH_CONFIGURATION + }); + + await expect(getReconcileAsync(components)()).resolves.toMatchObject({ + isFullInvalidation: true, + sequence: 2 + }); + expect(mapInvalidationsToOperationsAsync).not.toHaveBeenCalled(); + expect(invalidations.getSnapshot()).toMatchObject({ + changedPaths: [], + hasUnknownChanges: false + }); + await disposeComponentsAsync(components); + }); + + it('retains invalidations when a mapper returns an operation outside the graph', async () => { + const engine: ITestEngine = createTestEngine( + TEST_RUSH_CONFIGURATION.projects, + () => 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( + TEST_RUSH_CONFIGURATION.projects, + () => 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(TEST_RUSH_CONFIGURATION.projects, () => + 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( + TEST_RUSH_CONFIGURATION.projects, + () => 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( + TEST_RUSH_CONFIGURATION.projects, + () => 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 index 5f22668fa6..cf51deccfe 100644 --- a/libraries/rush-daemon/src/test/WorkspaceSession.test.ts +++ b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts @@ -213,4 +213,36 @@ describe(WorkspaceSession.name, () => { 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; + }); });