From ab2a20264c7fa2c6c566f94a79c7c45e94a3b1b5 Mon Sep 17 00:00:00 2001 From: mojaza Date: Tue, 18 Aug 2026 16:42:41 -0700 Subject: [PATCH 1/7] [rush-daemon] Add warm workspace session foundation Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...rm-workspace-session_2026-08-18-22-30.json | 11 + .../config/subspaces/default/pnpm-lock.yaml | 3 + common/reviews/api/rush-daemon.api.md | 122 +++++++++ libraries/rush-daemon/README.md | 10 +- libraries/rush-daemon/package.json | 1 + libraries/rush-daemon/src/RushDaemonHost.ts | 109 ++++++-- .../src/WorkspaceInvalidationTracker.ts | 86 ++++++ libraries/rush-daemon/src/WorkspaceSession.ts | 253 ++++++++++++++++++ .../src/WorkspaceSessionFileWatcher.ts | 112 ++++++++ .../src/WorkspaceSessionProvider.ts | 68 +++++ libraries/rush-daemon/src/index.ts | 15 ++ .../src/test/RushDaemonHost.test.ts | 109 +++++++- .../src/test/TestWorkspaceSession.ts | 49 ++++ .../src/test/WorkspaceSession.test.ts | 135 ++++++++++ .../src/test/WorkspaceSessionProvider.test.ts | 92 +++++++ 15 files changed, 1142 insertions(+), 33 deletions(-) create mode 100644 common/changes/@rushstack/rush-daemon/mojazayeri-warm-workspace-session_2026-08-18-22-30.json create mode 100644 libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts create mode 100644 libraries/rush-daemon/src/WorkspaceSession.ts create mode 100644 libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts create mode 100644 libraries/rush-daemon/src/WorkspaceSessionProvider.ts create mode 100644 libraries/rush-daemon/src/test/TestWorkspaceSession.ts create mode 100644 libraries/rush-daemon/src/test/WorkspaceSession.test.ts create mode 100644 libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts 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..cff3112f75 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -7,6 +7,23 @@ /// import type { IDaemonPaths } from '@rushstack/rush-daemon-transport'; +import type { IInputsSnapshot } from '@microsoft/rush-lib'; +import type { IOperationGraph } from '@microsoft/rush-lib'; +import { RushConfiguration } from '@microsoft/rush-lib'; +import type { RushSession } from '@microsoft/rush-lib'; + +// @beta +export type CreateWorkspaceSessionComponentsAsync = (options: ICreateWorkspaceSessionComponentsOptions) => Promise; + +// @beta +export interface ICreateWorkspaceSessionComponentsOptions { + // (undocumented) + readonly invalidations: WorkspaceInvalidationTracker; + // (undocumented) + readonly onError?: (error: Error) => void; + // (undocumented) + readonly rushConfiguration: RushConfiguration; +} // @public export interface IRequestLease { @@ -27,6 +44,7 @@ export interface IRequestSchedulerAcquireOptions { // @beta export interface IRushDaemonHostOptions { + readonly createWorkspaceSessionAsync?: WorkspaceSessionFactory; readonly daemonVersion: string; readonly onError?: (error: Error) => void; readonly repoRoot: string; @@ -40,6 +58,80 @@ export interface IRushDaemonServeOptions extends IRushDaemonHostOptions { readonly shutdownSignal?: AbortSignal; } +// @beta +export interface IWorkspaceInvalidationSnapshot { + readonly changedPaths: ReadonlyArray; + readonly hasUnknownChanges: boolean; + readonly isWatcherHealthy: boolean; + readonly sequence: number; +} + +// @beta +export interface IWorkspaceInvalidationWatcher { + // (undocumented) + disposeAsync(): Promise; + // (undocumented) + startAsync(onInvalidation: (changedPath?: string) => void): Promise; +} + +// @beta +export interface IWorkspaceSession { + // (undocumented) + disposeAsync(): Promise; + // (undocumented) + readonly inputsSnapshot: IInputsSnapshot | undefined; + // (undocumented) + readonly invalidations: WorkspaceInvalidationTracker; + // (undocumented) + readonly metadata: IWorkspaceSessionMetadata; + // (undocumented) + readonly operationGraph: IOperationGraph | undefined; + // (undocumented) + readonly rushConfiguration: RushConfiguration; + // (undocumented) + readonly rushSession: RushSession | undefined; +} + +// @beta +export interface IWorkspaceSessionComponents { + // (undocumented) + readonly disposeAsync?: () => Promise; + // (undocumented) + readonly inputsSnapshot?: IInputsSnapshot; + // (undocumented) + readonly operationGraph?: IOperationGraph; + // (undocumented) + readonly projectWatcher?: IWorkspaceInvalidationWatcher; + // (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; +} + // @public export enum RequestExclusivityClass { // (undocumented) @@ -77,6 +169,7 @@ export enum RequestSchedulerErrorCode { // @beta export class RushDaemonHost { closeAsync(): Promise; + getWorkspaceSessionAsync(): Promise; // (undocumented) readonly paths: IDaemonPaths; static startAsync(options: IRushDaemonHostOptions): Promise; @@ -85,6 +178,35 @@ export class RushDaemonHost { // @beta export function serveRushDaemonAsync(options: IRushDaemonServeOptions): Promise; +// @beta +export class WorkspaceInvalidationTracker { + acknowledgeThrough(sequence: number): void; + getSnapshot(): IWorkspaceInvalidationSnapshot; + invalidate(changedPath?: string): void; + markWatcherUnhealthy(): void; +} + +// @beta +export class WorkspaceSession implements IWorkspaceSession { + static createAsync(options: IWorkspaceSessionOptions): Promise; + disposeAsync(): Promise; + // (undocumented) + readonly inputsSnapshot: IInputsSnapshot | undefined; + // (undocumented) + readonly invalidations: WorkspaceInvalidationTracker; + // (undocumented) + readonly metadata: IWorkspaceSessionMetadata; + // (undocumented) + readonly operationGraph: IOperationGraph | undefined; + // (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..09c1e75ae3 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -1,8 +1,16 @@ # @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. 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. diff --git a/libraries/rush-daemon/package.json b/libraries/rush-daemon/package.json index 7daacb9ae7..08c566ea31 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..966f6b7da3 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.disposeAsync(); + } 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.disposeAsync(); + } 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/WorkspaceInvalidationTracker.ts b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts new file mode 100644 index 0000000000..8067373219 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts @@ -0,0 +1,86 @@ +// 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; +} + +/** + * Retains workspace invalidations until a future request explicitly acknowledges them. + * + * @beta + */ +export class WorkspaceInvalidationTracker { + private readonly _sequenceByPath: Map = new Map(); + private _latestSequence: number = 0; + private _unknownChangeSequence: number | undefined; + private _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 = sequence; + } else { + 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..af5a053a3d --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSession.ts @@ -0,0 +1,253 @@ +// 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'; + +/** + * 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 { + disposeAsync(): Promise; + 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 { + readonly disposeAsync?: () => Promise; + readonly inputsSnapshot?: IInputsSnapshot; + readonly operationGraph?: IOperationGraph; + readonly projectWatcher?: IWorkspaceInvalidationWatcher; + 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 { + readonly inputsSnapshot: IInputsSnapshot | undefined; + readonly invalidations: WorkspaceInvalidationTracker; + readonly metadata: IWorkspaceSessionMetadata; + readonly operationGraph: IOperationGraph | undefined; + readonly rushConfiguration: RushConfiguration; + readonly rushSession: RushSession | undefined; + disposeAsync(): Promise; +} + +/** + * Factory used by the daemon host to initialize its workspace session. + * + * @beta + */ +export type WorkspaceSessionFactory = (options: IWorkspaceSessionOptions) => Promise; + +/** + * A warm workspace session with client-independent invalidation tracking. + * + * @beta + */ +export class WorkspaceSession implements IWorkspaceSession { + private readonly _components: IWorkspaceSessionComponents; + private readonly _projectWatcher: IWorkspaceInvalidationWatcher; + private _disposePromise: Promise | undefined; + + public readonly inputsSnapshot: IInputsSnapshot | undefined; + 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, + projectWatcher: IWorkspaceInvalidationWatcher + ) { + this.rushConfiguration = rushConfiguration; + this.metadata = metadata; + this.invalidations = invalidations; + this._components = components; + this._projectWatcher = projectWatcher; + this.inputsSnapshot = components.inputsSnapshot; + this.operationGraph = components.operationGraph; + this.rushSession = components.rushSession; + } + + /** 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 + })) ?? {}; + let projectWatcher: IWorkspaceInvalidationWatcher | undefined = components.projectWatcher; + try { + const metadata: IWorkspaceSessionMetadata = createMetadata( + rushConfiguration, + options.rushVersion + ); + projectWatcher ??= new WorkspaceSessionFileWatcher({ + onError: (error: Error) => { + invalidations.markWatcherUnhealthy(); + options.onError?.(error); + }, + rushConfiguration + }); + const session: WorkspaceSession = new WorkspaceSession( + rushConfiguration, + metadata, + invalidations, + components, + projectWatcher + ); + await projectWatcher.startAsync((changedPath: string | undefined) => + invalidations.invalidate(changedPath) + ); + return session; + } catch (error) { + const cleanupErrors: unknown[] = []; + try { + await projectWatcher?.disposeAsync(); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + try { + await components.disposeAsync?.(); + } 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 disposeAsync(): Promise { + this._disposePromise ??= this._disposeOnceAsync(); + return this._disposePromise; + } + + private async _disposeOnceAsync(): Promise { + let watcherError: unknown; + try { + await this._projectWatcher.disposeAsync(); + } catch (error) { + watcherError = error; + } + + try { + await this._components.disposeAsync?.(); + } 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..5a55f8ac46 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts @@ -0,0 +1,112 @@ +// 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; +} + +interface IWatchPath { + readonly folderPath: string; + readonly recursive: boolean; +} + +export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatcher { + private readonly _onError: ((error: Error) => void) | undefined; + private readonly _watchPaths: ReadonlyArray; + private readonly _watchers: Set = new Set(); + private _onInvalidation: ((changedPath?: string) => void) | undefined; + private _disposed: boolean = false; + + public constructor(options: IWorkspaceSessionFileWatcherOptions) { + this._onError = options.onError; + 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; + try { + for (const watchPath of this._watchPaths) { + this._watchers.add(this._createWatcher(watchPath)); + } + } catch (error) { + await this.disposeAsync(); + throw error; + } + } + + public async disposeAsync(): 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; + } + + private _createWatcher(watchPath: IWatchPath): fs.FSWatcher { + const watcher: fs.FSWatcher = fs.watch( + 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 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(/[\\/]/) + .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..af05303cbe --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSessionProvider.ts @@ -0,0 +1,68 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IWorkspaceSession, + IWorkspaceSessionOptions, + WorkspaceSessionFactory +} from './WorkspaceSession'; + +export class WorkspaceSessionProvider { + private readonly _factory: WorkspaceSessionFactory; + private readonly _options: IWorkspaceSessionOptions; + private _initializationPromise: Promise | undefined; + private _session: IWorkspaceSession | undefined; + private _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 async disposeAsync(): Promise { + if (this._disposed) { + return; + } + this._disposed = true; + const session: IWorkspaceSession | undefined = + this._session ?? + (await this._initializationPromise?.then( + (initializedSession: IWorkspaceSession) => initializedSession, + () => undefined + )); + await session?.disposeAsync(); + this._session = undefined; + this._initializationPromise = undefined; + } + + private async _initializeAsync(): Promise { + const session: IWorkspaceSession = await this._factory(this._options); + if (this._disposed) { + await session.disposeAsync(); + 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..d0d438199b 100644 --- a/libraries/rush-daemon/src/index.ts +++ b/libraries/rush-daemon/src/index.ts @@ -13,3 +13,18 @@ export { } from './RequestScheduler'; export { RushDaemonHost, type IRushDaemonHostOptions } from './RushDaemonHost'; export { serveRushDaemonAsync, type IRushDaemonServeOptions } from './serveRushDaemon'; +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..148a7aff36 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,97 @@ 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).toEqual(['client', '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..59714e0be5 --- /dev/null +++ b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts @@ -0,0 +1,49 @@ +// 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 { + 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 { + private readonly _onDispose: (() => void) | 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?: () => void) { + this._onDispose = onDispose; + this.metadata = { + projectCount: 0, + projectNames: [], + repoRoot, + rushJsonFile: path.join(repoRoot, 'rush.json'), + rushVersion: '5.178.0' + }; + } + + public disposeAsync(): Promise { + this._onDispose?.(); + return Promise.resolve(); + } +} 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..867df4e4e0 --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts @@ -0,0 +1,135 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IWorkspaceInvalidationWatcher, + IWorkspaceSessionComponents +} from '../WorkspaceSession'; +import { WorkspaceSession } from '../WorkspaceSession'; +import type { IWorkspaceInvalidationSnapshot } from '../WorkspaceInvalidationTracker'; +import { TEST_REPO_ROOT } from './TestWorkspaceSession'; + +class TestInvalidationWatcher implements IWorkspaceInvalidationWatcher { + private readonly _events: string[]; + private _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 disposeAsync(): 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, + disposeAsync: () => { + events.push('components-dispose'); + return Promise.resolve(); + } + }); + } + }); + + 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() + ); + + 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: 3 + }); + + await session.disposeAsync(); + expect(events).toEqual(['watcher-start', 'watcher-dispose', 'components-dispose']); + }); + + it('does not allow a watcher error to be acknowledged as clean', async () => { + const session: WorkspaceSession = await WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: new TestInvalidationWatcher([]) + }) + }); + + 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.disposeAsync(); + }); + + it('disposes components when watcher startup fails', async () => { + const events: string[] = []; + const watcher: IWorkspaceInvalidationWatcher = { + startAsync: () => Promise.reject(new Error('watcher startup failed')), + disposeAsync: () => { + events.push('watcher-dispose'); + return Promise.resolve(); + } + }; + + await expect( + WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: watcher, + disposeAsync: () => { + events.push('components-dispose'); + return Promise.resolve(); + } + }) + }) + ).rejects.toThrow('watcher startup failed'); + expect(events).toEqual(['watcher-dispose', 'components-dispose']); + }); +}); 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..1ea715b335 --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts @@ -0,0 +1,92 @@ +// 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); + 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.disposeAsync(); + }); + + 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.disposeAsync(); + }); + + 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.disposeAsync(); + }); + + 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.disposeAsync(); + resolveFactory?.(session); + + await expect(initialization).rejects.toThrow('disposed during initialization'); + await disposal; + expect(disposalEvents).toEqual(['session']); + }); +}); From f916c98d5093491cdff99aacb6556b0cd6bb033c Mon Sep 17 00:00:00 2001 From: mojaza Date: Tue, 18 Aug 2026 16:58:13 -0700 Subject: [PATCH 2/7] [rush-daemon] Make lifecycle tests timing-safe Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- libraries/rush-daemon/src/test/RushDaemonHost.test.ts | 4 +++- .../rush-daemon/src/test/WorkspaceSessionProvider.test.ts | 2 ++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/libraries/rush-daemon/src/test/RushDaemonHost.test.ts b/libraries/rush-daemon/src/test/RushDaemonHost.test.ts index 148a7aff36..020d7e5ed3 100644 --- a/libraries/rush-daemon/src/test/RushDaemonHost.test.ts +++ b/libraries/rush-daemon/src/test/RushDaemonHost.test.ts @@ -163,7 +163,9 @@ describe(RushDaemonHost.name, () => { ); await host.closeAsync(); await closed; - expect(disposalEvents).toEqual(['client', 'workspace-session']); + 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' diff --git a/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts index 1ea715b335..a12f4395d6 100644 --- a/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts +++ b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts @@ -25,6 +25,7 @@ describe(WorkspaceSessionProvider.name, () => { const first: Promise = provider.getSessionAsync(); const second: Promise = provider.getSessionAsync(); expect(first).toBe(second); + await Promise.resolve(); expect(factoryCalls).toBe(1); resolveFactory?.(session); @@ -83,6 +84,7 @@ describe(WorkspaceSessionProvider.name, () => { const initialization: Promise = provider.getSessionAsync(); const disposal: Promise = provider.disposeAsync(); + await Promise.resolve(); resolveFactory?.(session); await expect(initialization).rejects.toThrow('disposed during initialization'); From 9feefa1eb95baec5bbbb5af97495c81040c1754e Mon Sep 17 00:00:00 2001 From: Mo Jazayeri Date: Wed, 19 Aug 2026 10:36:12 -0700 Subject: [PATCH 3/7] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts index 5a55f8ac46..f9d8d43e58 100644 --- a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts +++ b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts @@ -93,6 +93,9 @@ export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatche 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); } From 9a0945f64ddd0305c41106cea587d3584db394ef Mon Sep 17 00:00:00 2001 From: mojaza Date: Wed, 19 Aug 2026 11:04:15 -0700 Subject: [PATCH 4/7] [rush-daemon] Harden workspace invalidation tracking Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- libraries/rush-daemon/README.md | 8 ++-- .../src/WorkspaceInvalidationTracker.ts | 18 +++++-- libraries/rush-daemon/src/WorkspaceSession.ts | 2 + .../src/WorkspaceSessionFileWatcher.ts | 11 ++++- .../src/test/WorkspaceSession.test.ts | 37 ++++++++++++++- .../test/WorkspaceSessionFileWatcher.test.ts | 47 +++++++++++++++++++ 6 files changed, 115 insertions(+), 8 deletions(-) create mode 100644 libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index 09c1e75ae3..a2055f0727 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -10,7 +10,9 @@ for the nearest `rush.json`; it does not change the default behavior of `rush`, 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. 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 +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. diff --git a/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts index 8067373219..df9935cf20 100644 --- a/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts +++ b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts @@ -17,6 +17,8 @@ export interface IWorkspaceInvalidationSnapshot { readonly sequence: number; } +const MAX_TRACKED_CHANGED_PATHS: number = 10_000; + /** * Retains workspace invalidations until a future request explicitly acknowledges them. * @@ -31,11 +33,21 @@ export class WorkspaceInvalidationTracker { /** Records a path-specific or unknown workspace change. */ public invalidate(changedPath?: string): void { const sequence: number = ++this._latestSequence; - if (changedPath === undefined) { + if (changedPath === undefined || this._unknownChangeSequence !== undefined) { this._unknownChangeSequence = sequence; - } else { - this._sequenceByPath.set(changedPath, 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); } /** diff --git a/libraries/rush-daemon/src/WorkspaceSession.ts b/libraries/rush-daemon/src/WorkspaceSession.ts index af5a053a3d..871a34e83f 100644 --- a/libraries/rush-daemon/src/WorkspaceSession.ts +++ b/libraries/rush-daemon/src/WorkspaceSession.ts @@ -181,6 +181,8 @@ export class WorkspaceSession implements IWorkspaceSession { 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[] = []; diff --git a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts index f9d8d43e58..7290e69a87 100644 --- a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts +++ b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts @@ -12,6 +12,7 @@ import type { IWorkspaceInvalidationWatcher } from './WorkspaceSession'; export interface IWorkspaceSessionFileWatcherOptions { readonly onError?: (error: Error) => void; readonly rushConfiguration: RushConfiguration; + readonly watchFactory?: WorkspaceWatchFactory; } interface IWatchPath { @@ -19,8 +20,15 @@ interface IWatchPath { readonly recursive: boolean; } +type WorkspaceWatchFactory = ( + folderPath: string, + options: { encoding: 'utf8'; recursive: boolean }, + listener: fs.WatchListener +) => fs.FSWatcher; + export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatcher { private readonly _onError: ((error: Error) => void) | undefined; + private readonly _watchFactory: WorkspaceWatchFactory; private readonly _watchPaths: ReadonlyArray; private readonly _watchers: Set = new Set(); private _onInvalidation: ((changedPath?: string) => void) | undefined; @@ -28,6 +36,7 @@ export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatche public constructor(options: IWorkspaceSessionFileWatcherOptions) { this._onError = options.onError; + this._watchFactory = options.watchFactory ?? fs.watch; this._watchPaths = getWatchPaths(options.rushConfiguration); } @@ -66,7 +75,7 @@ export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatche } private _createWatcher(watchPath: IWatchPath): fs.FSWatcher { - const watcher: fs.FSWatcher = fs.watch( + const watcher: fs.FSWatcher = this._watchFactory( watchPath.folderPath, { encoding: 'utf8', recursive: watchPath.recursive }, (eventType: string, filename: string | null) => { diff --git a/libraries/rush-daemon/src/test/WorkspaceSession.test.ts b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts index 867df4e4e0..776b8bfd5d 100644 --- a/libraries/rush-daemon/src/test/WorkspaceSession.test.ts +++ b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts @@ -7,6 +7,7 @@ import type { } 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 { @@ -68,6 +69,15 @@ describe(WorkspaceSession.name, () => { 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'); @@ -78,7 +88,7 @@ describe(WorkspaceSession.name, () => { changedPaths: ['packages/a/src/index.ts'], hasUnknownChanges: true, isWatcherHealthy: true, - sequence: 3 + sequence: 4 }); await session.disposeAsync(); @@ -106,6 +116,31 @@ describe(WorkspaceSession.name, () => { await session.disposeAsync(); }); + 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 = { 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..c7ce709323 --- /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.disposeAsync(); + }); +}); From 80ab0c4ec2b62a77a0429217b903e5450f021e85 Mon Sep 17 00:00:00 2001 From: mojaza Date: Wed, 19 Aug 2026 20:25:05 -0700 Subject: [PATCH 5/7] [rush-daemon] Adopt async disposal contracts Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- common/reviews/api/rush-daemon.api.md | 14 +-- libraries/rush-daemon/src/RushDaemonHost.ts | 4 +- libraries/rush-daemon/src/WorkspaceSession.ts | 41 ++++----- .../src/WorkspaceSessionFileWatcher.ts | 57 ++++++------- .../src/WorkspaceSessionProvider.ts | 85 +++++++++++-------- .../src/test/TestWorkspaceSession.ts | 11 ++- .../src/test/WorkspaceSession.test.ts | 44 ++++++++-- .../test/WorkspaceSessionFileWatcher.test.ts | 2 +- .../src/test/WorkspaceSessionProvider.test.ts | 62 +++++++++++++- 9 files changed, 204 insertions(+), 116 deletions(-) diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index cff3112f75..863f0f8065 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -67,17 +67,13 @@ export interface IWorkspaceInvalidationSnapshot { } // @beta -export interface IWorkspaceInvalidationWatcher { - // (undocumented) - disposeAsync(): Promise; +export interface IWorkspaceInvalidationWatcher extends AsyncDisposable { // (undocumented) startAsync(onInvalidation: (changedPath?: string) => void): Promise; } // @beta -export interface IWorkspaceSession { - // (undocumented) - disposeAsync(): Promise; +export interface IWorkspaceSession extends AsyncDisposable { // (undocumented) readonly inputsSnapshot: IInputsSnapshot | undefined; // (undocumented) @@ -93,9 +89,7 @@ export interface IWorkspaceSession { } // @beta -export interface IWorkspaceSessionComponents { - // (undocumented) - readonly disposeAsync?: () => Promise; +export interface IWorkspaceSessionComponents extends AsyncDisposable { // (undocumented) readonly inputsSnapshot?: IInputsSnapshot; // (undocumented) @@ -188,8 +182,8 @@ export class WorkspaceInvalidationTracker { // @beta export class WorkspaceSession implements IWorkspaceSession { + [Symbol.asyncDispose](): Promise; static createAsync(options: IWorkspaceSessionOptions): Promise; - disposeAsync(): Promise; // (undocumented) readonly inputsSnapshot: IInputsSnapshot | undefined; // (undocumented) diff --git a/libraries/rush-daemon/src/RushDaemonHost.ts b/libraries/rush-daemon/src/RushDaemonHost.ts index 966f6b7da3..15cc941fc3 100644 --- a/libraries/rush-daemon/src/RushDaemonHost.ts +++ b/libraries/rush-daemon/src/RushDaemonHost.ts @@ -112,7 +112,7 @@ export class RushDaemonHost { }); } catch (error) { try { - await workspaceSessionProvider.disposeAsync(); + await workspaceSessionProvider[Symbol.asyncDispose](); } catch (cleanupError) { throw new AggregateError( [error, cleanupError], @@ -157,7 +157,7 @@ export class RushDaemonHost { errors.push(error); } try { - await this._workspaceSessionProvider.disposeAsync(); + await this._workspaceSessionProvider[Symbol.asyncDispose](); } catch (error) { errors.push(error); } diff --git a/libraries/rush-daemon/src/WorkspaceSession.ts b/libraries/rush-daemon/src/WorkspaceSession.ts index 871a34e83f..bd6656e216 100644 --- a/libraries/rush-daemon/src/WorkspaceSession.ts +++ b/libraries/rush-daemon/src/WorkspaceSession.ts @@ -31,8 +31,7 @@ export interface IWorkspaceSessionMetadata { * * @beta */ -export interface IWorkspaceInvalidationWatcher { - disposeAsync(): Promise; +export interface IWorkspaceInvalidationWatcher extends AsyncDisposable { startAsync(onInvalidation: (changedPath?: string) => void): Promise; } @@ -45,8 +44,7 @@ export interface IWorkspaceInvalidationWatcher { * * @beta */ -export interface IWorkspaceSessionComponents { - readonly disposeAsync?: () => Promise; +export interface IWorkspaceSessionComponents extends AsyncDisposable { readonly inputsSnapshot?: IInputsSnapshot; readonly operationGraph?: IOperationGraph; readonly projectWatcher?: IWorkspaceInvalidationWatcher; @@ -90,14 +88,13 @@ export interface IWorkspaceSessionOptions { * * @beta */ -export interface IWorkspaceSession { +export interface IWorkspaceSession extends AsyncDisposable { readonly inputsSnapshot: IInputsSnapshot | undefined; readonly invalidations: WorkspaceInvalidationTracker; readonly metadata: IWorkspaceSessionMetadata; readonly operationGraph: IOperationGraph | undefined; readonly rushConfiguration: RushConfiguration; readonly rushSession: RushSession | undefined; - disposeAsync(): Promise; } /** @@ -107,15 +104,19 @@ export interface IWorkspaceSession { */ 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 { - private readonly _components: IWorkspaceSessionComponents; - private readonly _projectWatcher: IWorkspaceInvalidationWatcher; - private _disposePromise: Promise | undefined; + readonly #components: IWorkspaceSessionComponents; + readonly #projectWatcher: IWorkspaceInvalidationWatcher; + #disposePromise: Promise | undefined; public readonly inputsSnapshot: IInputsSnapshot | undefined; public readonly invalidations: WorkspaceInvalidationTracker; @@ -134,8 +135,8 @@ export class WorkspaceSession implements IWorkspaceSession { this.rushConfiguration = rushConfiguration; this.metadata = metadata; this.invalidations = invalidations; - this._components = components; - this._projectWatcher = projectWatcher; + this.#components = components; + this.#projectWatcher = projectWatcher; this.inputsSnapshot = components.inputsSnapshot; this.operationGraph = components.operationGraph; this.rushSession = components.rushSession; @@ -157,7 +158,7 @@ export class WorkspaceSession implements IWorkspaceSession { invalidations, onError: options.onError, rushConfiguration - })) ?? {}; + })) ?? EMPTY_WORKSPACE_SESSION_COMPONENTS; let projectWatcher: IWorkspaceInvalidationWatcher | undefined = components.projectWatcher; try { const metadata: IWorkspaceSessionMetadata = createMetadata( @@ -187,12 +188,12 @@ export class WorkspaceSession implements IWorkspaceSession { } catch (error) { const cleanupErrors: unknown[] = []; try { - await projectWatcher?.disposeAsync(); + await projectWatcher?.[Symbol.asyncDispose](); } catch (cleanupError) { cleanupErrors.push(cleanupError); } try { - await components.disposeAsync?.(); + await components[Symbol.asyncDispose](); } catch (cleanupError) { cleanupErrors.push(cleanupError); } @@ -207,21 +208,21 @@ export class WorkspaceSession implements IWorkspaceSession { } /** Stops invalidation tracking and disposes injected engine resources. */ - public disposeAsync(): Promise { - this._disposePromise ??= this._disposeOnceAsync(); - return this._disposePromise; + public [Symbol.asyncDispose](): Promise { + this.#disposePromise ??= this.#disposeOnceAsync(); + return this.#disposePromise; } - private async _disposeOnceAsync(): Promise { + async #disposeOnceAsync(): Promise { let watcherError: unknown; try { - await this._projectWatcher.disposeAsync(); + await this.#projectWatcher[Symbol.asyncDispose](); } catch (error) { watcherError = error; } try { - await this._components.disposeAsync?.(); + await this.#components[Symbol.asyncDispose](); } catch (componentError) { if (watcherError !== undefined) { throw new AggregateError( diff --git a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts index 7290e69a87..3977c82075 100644 --- a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts +++ b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts @@ -27,62 +27,57 @@ type WorkspaceWatchFactory = ( ) => fs.FSWatcher; export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatcher { - private readonly _onError: ((error: Error) => void) | undefined; - private readonly _watchFactory: WorkspaceWatchFactory; - private readonly _watchPaths: ReadonlyArray; - private readonly _watchers: Set = new Set(); - private _onInvalidation: ((changedPath?: string) => void) | undefined; - private _disposed: boolean = false; + 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); + 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) { + if (this.#disposed) { throw new Error('The workspace watcher has already been disposed.'); } - if (this._onInvalidation) { + if (this.#onInvalidation) { throw new Error('The workspace watcher has already been started.'); } - this._onInvalidation = onInvalidation; - try { - for (const watchPath of this._watchPaths) { - this._watchers.add(this._createWatcher(watchPath)); - } - } catch (error) { - await this.disposeAsync(); - throw error; + this.#onInvalidation = onInvalidation; + for (const watchPath of this.#watchPaths) { + this.#watchers.add(this.#createWatcher(watchPath)); } } - public async disposeAsync(): Promise { - if (this._disposed) { + public async [Symbol.asyncDispose](): Promise { + if (this.#disposed) { return; } - this._disposed = true; + this.#disposed = true; const closePromises: Promise[] = []; - for (const watcher of this._watchers) { + for (const watcher of this.#watchers) { closePromises.push(once(watcher, 'close')); watcher.close(); } await Promise.all(closePromises); - this._watchers.clear(); - this._onInvalidation = undefined; + this.#watchers.clear(); + this.#onInvalidation = undefined; } - private _createWatcher(watchPath: IWatchPath): fs.FSWatcher { - const watcher: fs.FSWatcher = this._watchFactory( + #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?.( + this.#onInvalidation?.( changedFilename === undefined ? undefined : path.resolve(watchPath.folderPath, changedFilename) @@ -91,10 +86,10 @@ export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatche } ); watcher.on('error', (error: Error) => { - this._onInvalidation?.(); - this._onError?.(error); + this.#onInvalidation?.(); + this.#onError?.(error); }); - watcher.once('close', () => this._watchers.delete(watcher)); + watcher.once('close', () => this.#watchers.delete(watcher)); watcher.unref(); return watcher; } diff --git a/libraries/rush-daemon/src/WorkspaceSessionProvider.ts b/libraries/rush-daemon/src/WorkspaceSessionProvider.ts index af05303cbe..85ba8eba51 100644 --- a/libraries/rush-daemon/src/WorkspaceSessionProvider.ts +++ b/libraries/rush-daemon/src/WorkspaceSessionProvider.ts @@ -7,62 +7,77 @@ import type { WorkspaceSessionFactory } from './WorkspaceSession'; -export class WorkspaceSessionProvider { - private readonly _factory: WorkspaceSessionFactory; - private readonly _options: IWorkspaceSessionOptions; - private _initializationPromise: Promise | undefined; - private _session: IWorkspaceSession | undefined; - private _disposed: boolean = false; +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; + this.#factory = factory; + this.#options = options; } public getSessionAsync(): Promise { - if (this._disposed) { + 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.#session) { + return Promise.resolve(this.#session); } - if (!this._initializationPromise) { + if (!this.#initializationPromise) { const initializationPromise: Promise = Promise.resolve().then(() => - this._initializeAsync() + this.#initializeAsync() ); - this._initializationPromise = initializationPromise; + this.#initializationPromise = initializationPromise; void initializationPromise.catch(() => { - if (this._initializationPromise === initializationPromise) { - this._initializationPromise = undefined; + if (this.#initializationPromise === initializationPromise) { + this.#initializationPromise = undefined; } }); } - return this._initializationPromise; + return this.#initializationPromise; } - public async disposeAsync(): Promise { - if (this._disposed) { - return; + 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; } - this._disposed = true; - const session: IWorkspaceSession | undefined = - this._session ?? - (await this._initializationPromise?.then( - (initializedSession: IWorkspaceSession) => initializedSession, - () => undefined - )); - await session?.disposeAsync(); - this._session = undefined; - this._initializationPromise = undefined; } - private async _initializeAsync(): Promise { - const session: IWorkspaceSession = await this._factory(this._options); - if (this._disposed) { - await session.disposeAsync(); + 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; + this.#session = session; return session; } } diff --git a/libraries/rush-daemon/src/test/TestWorkspaceSession.ts b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts index 59714e0be5..fb609e0c73 100644 --- a/libraries/rush-daemon/src/test/TestWorkspaceSession.ts +++ b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts @@ -22,7 +22,7 @@ export const TEST_RUSH_CONFIGURATION: RushConfiguration = RushConfiguration.load ); export class TestWorkspaceSession implements IWorkspaceSession { - private readonly _onDispose: (() => void) | undefined; + readonly #onDispose: (() => unknown) | undefined; public readonly inputsSnapshot: IInputsSnapshot | undefined; public readonly invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); @@ -31,8 +31,8 @@ export class TestWorkspaceSession implements IWorkspaceSession { public readonly rushConfiguration: RushConfiguration = TEST_RUSH_CONFIGURATION; public readonly rushSession: RushSession | undefined; - public constructor(repoRoot: string, onDispose?: () => void) { - this._onDispose = onDispose; + public constructor(repoRoot: string, onDispose?: () => unknown) { + this.#onDispose = onDispose; this.metadata = { projectCount: 0, projectNames: [], @@ -42,8 +42,7 @@ export class TestWorkspaceSession implements IWorkspaceSession { }; } - public disposeAsync(): Promise { - this._onDispose?.(); - return Promise.resolve(); + public async [Symbol.asyncDispose](): Promise { + await this.#onDispose?.(); } } diff --git a/libraries/rush-daemon/src/test/WorkspaceSession.test.ts b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts index 776b8bfd5d..aa4bda35b8 100644 --- a/libraries/rush-daemon/src/test/WorkspaceSession.test.ts +++ b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts @@ -31,7 +31,7 @@ class TestInvalidationWatcher implements IWorkspaceInvalidationWatcher { this._onInvalidation(changedPath); } - public disposeAsync(): Promise { + public [Symbol.asyncDispose](): Promise { this._events.push('watcher-dispose'); this._onInvalidation = undefined; return Promise.resolve(); @@ -50,7 +50,7 @@ describe(WorkspaceSession.name, () => { componentFactoryCalls++; return Promise.resolve({ projectWatcher: watcher, - disposeAsync: () => { + [Symbol.asyncDispose]: () => { events.push('components-dispose'); return Promise.resolve(); } @@ -91,7 +91,7 @@ describe(WorkspaceSession.name, () => { sequence: 4 }); - await session.disposeAsync(); + await session[Symbol.asyncDispose](); expect(events).toEqual(['watcher-start', 'watcher-dispose', 'components-dispose']); }); @@ -101,7 +101,8 @@ describe(WorkspaceSession.name, () => { rushVersion: '5.178.0', createComponentsAsync: () => Promise.resolve({ - projectWatcher: new TestInvalidationWatcher([]) + projectWatcher: new TestInvalidationWatcher([]), + [Symbol.asyncDispose]: () => Promise.resolve() }) }); @@ -113,7 +114,7 @@ describe(WorkspaceSession.name, () => { hasUnknownChanges: true, isWatcherHealthy: false }); - await session.disposeAsync(); + await session[Symbol.asyncDispose](); }); it('compacts excessive path changes into an unknown invalidation', () => { @@ -145,7 +146,7 @@ describe(WorkspaceSession.name, () => { const events: string[] = []; const watcher: IWorkspaceInvalidationWatcher = { startAsync: () => Promise.reject(new Error('watcher startup failed')), - disposeAsync: () => { + [Symbol.asyncDispose]: () => { events.push('watcher-dispose'); return Promise.resolve(); } @@ -158,7 +159,7 @@ describe(WorkspaceSession.name, () => { createComponentsAsync: () => Promise.resolve({ projectWatcher: watcher, - disposeAsync: () => { + [Symbol.asyncDispose]: () => { events.push('components-dispose'); return Promise.resolve(); } @@ -167,4 +168,33 @@ describe(WorkspaceSession.name, () => { ).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]: () => Promise.reject(new Error('component cleanup failed')) + }) + }); + } catch (error) { + thrownError = error; + } + + expect(thrownError).toBeInstanceOf(AggregateError); + expect((thrownError as AggregateError).errors).toEqual([ + new Error('watcher startup failed'), + new Error('watcher cleanup failed'), + new Error('component cleanup failed') + ]); + }); }); diff --git a/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts b/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts index c7ce709323..1d8a364dc8 100644 --- a/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts +++ b/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts @@ -42,6 +42,6 @@ describe(WorkspaceSessionFileWatcher.name, () => { expect.arrayContaining(subspaceConfigFolders) ); - await watcher.disposeAsync(); + await watcher[Symbol.asyncDispose](); }); }); diff --git a/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts index a12f4395d6..a5328edaac 100644 --- a/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts +++ b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts @@ -32,7 +32,7 @@ describe(WorkspaceSessionProvider.name, () => { await expect(first).resolves.toBe(session); await expect(provider.getSessionAsync()).resolves.toBe(session); expect(factoryCalls).toBe(1); - await provider.disposeAsync(); + await provider[Symbol.asyncDispose](); }); it('clears a failed initialization so a later attempt can retry', async () => { @@ -48,7 +48,7 @@ describe(WorkspaceSessionProvider.name, () => { await expect(provider.getSessionAsync()).rejects.toThrow('initialization failed'); await expect(provider.getSessionAsync()).resolves.toBe(session); expect(factoryCalls).toBe(2); - await provider.disposeAsync(); + await provider[Symbol.asyncDispose](); }); it('clears a synchronously thrown initialization so a later attempt can retry', async () => { @@ -65,7 +65,7 @@ describe(WorkspaceSessionProvider.name, () => { await expect(provider.getSessionAsync()).rejects.toThrow('synchronous initialization failed'); await expect(provider.getSessionAsync()).resolves.toBe(session); expect(factoryCalls).toBe(2); - await provider.disposeAsync(); + await provider[Symbol.asyncDispose](); }); it('disposes a session that finishes initializing during shutdown', async () => { @@ -83,7 +83,7 @@ describe(WorkspaceSessionProvider.name, () => { ); const initialization: Promise = provider.getSessionAsync(); - const disposal: Promise = provider.disposeAsync(); + const disposal: Promise = provider[Symbol.asyncDispose](); await Promise.resolve(); resolveFactory?.(session); @@ -91,4 +91,58 @@ describe(WorkspaceSessionProvider.name, () => { 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]); + }); }); From 7f52c3486d32820e011fb497c3dd889303a42217 Mon Sep 17 00:00:00 2001 From: mojaza Date: Wed, 19 Aug 2026 20:59:28 -0700 Subject: [PATCH 6/7] chore: rerun CI Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> From a0432d8ffe0a083e44406a933eec2a033d7b15ad Mon Sep 17 00:00:00 2001 From: mojaza Date: Thu, 20 Aug 2026 09:47:52 -0700 Subject: [PATCH 7/7] [rush-daemon] Clarify workspace watcher ownership Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- common/reviews/api/rush-daemon.api.md | 1 - .../src/WorkspaceInvalidationTracker.ts | 50 +++++++++--------- libraries/rush-daemon/src/WorkspaceSession.ts | 37 ++++++++----- .../src/WorkspaceSessionFileWatcher.ts | 4 +- .../src/test/WorkspaceSession.test.ts | 52 ++++++++++++------- 5 files changed, 86 insertions(+), 58 deletions(-) diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index 863f0f8065..aa8bb2d015 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -94,7 +94,6 @@ export interface IWorkspaceSessionComponents extends AsyncDisposable { readonly inputsSnapshot?: IInputsSnapshot; // (undocumented) readonly operationGraph?: IOperationGraph; - // (undocumented) readonly projectWatcher?: IWorkspaceInvalidationWatcher; // (undocumented) readonly rushSession?: RushSession; diff --git a/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts index df9935cf20..ee67d5afa1 100644 --- a/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts +++ b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts @@ -25,29 +25,29 @@ const MAX_TRACKED_CHANGED_PATHS: number = 10_000; * @beta */ export class WorkspaceInvalidationTracker { - private readonly _sequenceByPath: Map = new Map(); - private _latestSequence: number = 0; - private _unknownChangeSequence: number | undefined; - private _watcherHealthy: boolean = true; + 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; + 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.has(changedPath) && + this.#sequenceByPath.size >= MAX_TRACKED_CHANGED_PATHS ) { - this._sequenceByPath.clear(); - this._unknownChangeSequence = sequence; + this.#sequenceByPath.clear(); + this.#unknownChangeSequence = sequence; return; } - this._sequenceByPath.set(changedPath, sequence); + this.#sequenceByPath.set(changedPath, sequence); } /** @@ -56,8 +56,8 @@ export class WorkspaceInvalidationTracker { * Unknown invalidation remains pending so consumers cannot mistake the workspace for clean. */ public markWatcherUnhealthy(): void { - if (this._watcherHealthy) { - this._watcherHealthy = false; + if (this.#watcherHealthy) { + this.#watcherHealthy = false; this.invalidate(); } } @@ -65,10 +65,10 @@ export class WorkspaceInvalidationTracker { /** 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 + changedPaths: Array.from(this.#sequenceByPath.keys()).sort(), + hasUnknownChanges: this.#unknownChangeSequence !== undefined, + isWatcherHealthy: this.#watcherHealthy, + sequence: this.#latestSequence }; } @@ -78,21 +78,21 @@ export class WorkspaceInvalidationTracker { * 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) { + if (!Number.isSafeInteger(sequence) || sequence < 0 || sequence > this.#latestSequence) { throw new RangeError(`Invalid workspace invalidation sequence: ${sequence}`); } - for (const [changedPath, pathSequence] of this._sequenceByPath) { + for (const [changedPath, pathSequence] of this.#sequenceByPath) { if (pathSequence <= sequence) { - this._sequenceByPath.delete(changedPath); + this.#sequenceByPath.delete(changedPath); } } if ( - this._watcherHealthy && - this._unknownChangeSequence !== undefined && - this._unknownChangeSequence <= sequence + this.#watcherHealthy && + this.#unknownChangeSequence !== undefined && + this.#unknownChangeSequence <= sequence ) { - this._unknownChangeSequence = undefined; + this.#unknownChangeSequence = undefined; } } } diff --git a/libraries/rush-daemon/src/WorkspaceSession.ts b/libraries/rush-daemon/src/WorkspaceSession.ts index bd6656e216..8abe498e1f 100644 --- a/libraries/rush-daemon/src/WorkspaceSession.ts +++ b/libraries/rush-daemon/src/WorkspaceSession.ts @@ -47,6 +47,13 @@ export interface IWorkspaceInvalidationWatcher extends AsyncDisposable { export interface IWorkspaceSessionComponents extends AsyncDisposable { 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 rushSession?: RushSession; } @@ -115,7 +122,7 @@ const EMPTY_WORKSPACE_SESSION_COMPONENTS: IWorkspaceSessionComponents = { */ export class WorkspaceSession implements IWorkspaceSession { readonly #components: IWorkspaceSessionComponents; - readonly #projectWatcher: IWorkspaceInvalidationWatcher; + readonly #sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined; #disposePromise: Promise | undefined; public readonly inputsSnapshot: IInputsSnapshot | undefined; @@ -130,13 +137,13 @@ export class WorkspaceSession implements IWorkspaceSession { metadata: IWorkspaceSessionMetadata, invalidations: WorkspaceInvalidationTracker, components: IWorkspaceSessionComponents, - projectWatcher: IWorkspaceInvalidationWatcher + sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined ) { this.rushConfiguration = rushConfiguration; this.metadata = metadata; this.invalidations = invalidations; this.#components = components; - this.#projectWatcher = projectWatcher; + this.#sessionOwnedProjectWatcher = sessionOwnedProjectWatcher; this.inputsSnapshot = components.inputsSnapshot; this.operationGraph = components.operationGraph; this.rushSession = components.rushSession; @@ -160,24 +167,28 @@ export class WorkspaceSession implements IWorkspaceSession { rushConfiguration })) ?? EMPTY_WORKSPACE_SESSION_COMPONENTS; let projectWatcher: IWorkspaceInvalidationWatcher | undefined = components.projectWatcher; + let sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined; try { const metadata: IWorkspaceSessionMetadata = createMetadata( rushConfiguration, options.rushVersion ); - projectWatcher ??= new WorkspaceSessionFileWatcher({ - onError: (error: Error) => { - invalidations.markWatcherUnhealthy(); - options.onError?.(error); - }, - rushConfiguration - }); + 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, - projectWatcher + sessionOwnedProjectWatcher ); await projectWatcher.startAsync((changedPath: string | undefined) => invalidations.invalidate(changedPath) @@ -188,7 +199,7 @@ export class WorkspaceSession implements IWorkspaceSession { } catch (error) { const cleanupErrors: unknown[] = []; try { - await projectWatcher?.[Symbol.asyncDispose](); + await sessionOwnedProjectWatcher?.[Symbol.asyncDispose](); } catch (cleanupError) { cleanupErrors.push(cleanupError); } @@ -216,7 +227,7 @@ export class WorkspaceSession implements IWorkspaceSession { async #disposeOnceAsync(): Promise { let watcherError: unknown; try { - await this.#projectWatcher[Symbol.asyncDispose](); + await this.#sessionOwnedProjectWatcher?.[Symbol.asyncDispose](); } catch (error) { watcherError = error; } diff --git a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts index 3977c82075..acee521b21 100644 --- a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts +++ b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts @@ -26,6 +26,8 @@ type WorkspaceWatchFactory = ( 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; @@ -114,6 +116,6 @@ function isIgnoredPath(filename: string | undefined): boolean { return false; } return filename - .split(/[\\/]/) + .split(PATH_SEGMENT_SEPARATOR_REGEXP) .some((segment: string) => segment === '.git' || segment === 'node_modules'); } diff --git a/libraries/rush-daemon/src/test/WorkspaceSession.test.ts b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts index aa4bda35b8..5f22668fa6 100644 --- a/libraries/rush-daemon/src/test/WorkspaceSession.test.ts +++ b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts @@ -11,29 +11,29 @@ import { WorkspaceInvalidationTracker } from '../WorkspaceInvalidationTracker'; import { TEST_REPO_ROOT } from './TestWorkspaceSession'; class TestInvalidationWatcher implements IWorkspaceInvalidationWatcher { - private readonly _events: string[]; - private _onInvalidation: ((changedPath?: string) => void) | undefined; + readonly #events: string[]; + #onInvalidation: ((changedPath?: string) => void) | undefined; public constructor(events: string[]) { - this._events = events; + this.#events = events; } public startAsync(onInvalidation: (changedPath?: string) => void): Promise { - this._events.push('watcher-start'); - this._onInvalidation = onInvalidation; + this.#events.push('watcher-start'); + this.#onInvalidation = onInvalidation; return Promise.resolve(); } public invalidate(changedPath?: string): void { - if (!this._onInvalidation) { + if (!this.#onInvalidation) { throw new Error('The test watcher is not running.'); } - this._onInvalidation(changedPath); + this.#onInvalidation(changedPath); } public [Symbol.asyncDispose](): Promise { - this._events.push('watcher-dispose'); - this._onInvalidation = undefined; + this.#events.push('watcher-dispose'); + this.#onInvalidation = undefined; return Promise.resolve(); } } @@ -50,9 +50,9 @@ describe(WorkspaceSession.name, () => { componentFactoryCalls++; return Promise.resolve({ projectWatcher: watcher, - [Symbol.asyncDispose]: () => { + [Symbol.asyncDispose]: async () => { + await watcher[Symbol.asyncDispose](); events.push('components-dispose'); - return Promise.resolve(); } }); } @@ -93,16 +93,19 @@ describe(WorkspaceSession.name, () => { 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: new TestInvalidationWatcher([]), - [Symbol.asyncDispose]: () => Promise.resolve() + projectWatcher: watcher, + [Symbol.asyncDispose]: () => watcher[Symbol.asyncDispose]() }) }); @@ -159,9 +162,9 @@ describe(WorkspaceSession.name, () => { createComponentsAsync: () => Promise.resolve({ projectWatcher: watcher, - [Symbol.asyncDispose]: () => { + [Symbol.asyncDispose]: async () => { + await watcher[Symbol.asyncDispose](); events.push('components-dispose'); - return Promise.resolve(); } }) }) @@ -183,7 +186,18 @@ describe(WorkspaceSession.name, () => { createComponentsAsync: () => Promise.resolve({ projectWatcher: watcher, - [Symbol.asyncDispose]: () => Promise.reject(new Error('component cleanup failed')) + [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) { @@ -191,8 +205,10 @@ describe(WorkspaceSession.name, () => { } expect(thrownError).toBeInstanceOf(AggregateError); - expect((thrownError as AggregateError).errors).toEqual([ - new Error('watcher startup failed'), + 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') ]);