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..aa8bb2d015 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,73 @@ 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 extends AsyncDisposable { + // (undocumented) + startAsync(onInvalidation: (changedPath?: string) => void): Promise; +} + +// @beta +export interface IWorkspaceSession extends AsyncDisposable { + // (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 extends AsyncDisposable { + // (undocumented) + readonly inputsSnapshot?: IInputsSnapshot; + // (undocumented) + readonly operationGraph?: IOperationGraph; + 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 +162,7 @@ export enum RequestSchedulerErrorCode { // @beta export class RushDaemonHost { closeAsync(): Promise; + getWorkspaceSessionAsync(): Promise; // (undocumented) readonly paths: IDaemonPaths; static startAsync(options: IRushDaemonHostOptions): Promise; @@ -85,6 +171,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 { + [Symbol.asyncDispose](): Promise; + static createAsync(options: IWorkspaceSessionOptions): 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..a2055f0727 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -1,8 +1,18 @@ # @rushstack/rush-daemon The long-lived Rush workspace daemon host, including workspace-keyed listener bootstrap, -protocol handshake and liveness control, and explicit serve/shutdown lifecycle APIs. +protocol handshake and liveness control, a warm `WorkspaceSession`, and explicit +serve/shutdown lifecycle APIs. The package provides an opt-in `rushd` executable. Run it from a Rush workspace to start the host for the nearest `rush.json`; it does not change the default behavior of `rush`, `rushx`, or `rush-pnpm`. + +The host loads `RushConfiguration` once before signaling readiness and keeps a headless file watcher +active for the daemon lifetime. Its invalidation tracker retains changes while no clients are +connected so a later request can reconcile them. The tracker starts with a conservative unknown +invalidation covering session startup, and excessive distinct paths are compacted into the same +full-workspace signal. 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 f035489be6..5c4c1c33cf 100644 --- a/libraries/rush-daemon/package.json +++ b/libraries/rush-daemon/package.json @@ -45,6 +45,7 @@ "_phase:test": "heft run --only test -- --clean" }, "dependencies": { + "@microsoft/rush-lib": "workspace:*", "@rushstack/node-core-library": "workspace:*", "@rushstack/rush-daemon-protocol": "workspace:*", "@rushstack/rush-daemon-transport": "workspace:*" diff --git a/libraries/rush-daemon/src/RushDaemonHost.ts b/libraries/rush-daemon/src/RushDaemonHost.ts index f3a9cd4251..15cc941fc3 100644 --- a/libraries/rush-daemon/src/RushDaemonHost.ts +++ b/libraries/rush-daemon/src/RushDaemonHost.ts @@ -15,6 +15,9 @@ import type { } from '@rushstack/rush-daemon-transport'; import { DaemonControlSession } from './DaemonControlSession'; +import { WorkspaceSession } from './WorkspaceSession'; +import type { IWorkspaceSession, WorkspaceSessionFactory } from './WorkspaceSession'; +import { WorkspaceSessionProvider } from './WorkspaceSessionProvider'; /** * Options for starting one workspace daemon host. @@ -22,6 +25,8 @@ import { DaemonControlSession } from './DaemonControlSession'; * @beta */ export interface IRushDaemonHostOptions { + /** Overrides workspace session construction for engine integration or testing. */ + readonly createWorkspaceSessionAsync?: WorkspaceSessionFactory; /** The daemon implementation version reported by `pong`. */ readonly daemonVersion: string; /** Reports connection-level failures. */ @@ -42,6 +47,7 @@ export interface IRushDaemonHostOptions { export class RushDaemonHost { private readonly _listener: DaemonFrameListener; private readonly _sessions: Set; + private readonly _workspaceSessionProvider: WorkspaceSessionProvider; private readonly _lifecycle: { closing: boolean }; public readonly paths: IDaemonPaths; private _closePromise: Promise | undefined; @@ -50,12 +56,14 @@ export class RushDaemonHost { listener: DaemonFrameListener, paths: IDaemonPaths, sessions: Set, - lifecycle: { closing: boolean } + lifecycle: { closing: boolean }, + workspaceSessionProvider: WorkspaceSessionProvider ) { this._listener = listener; this.paths = paths; this._sessions = sessions; this._lifecycle = lifecycle; + this._workspaceSessionProvider = workspaceSessionProvider; } /** Resolves only after the transport is bound and its lockfile has been written. */ @@ -69,29 +77,62 @@ export class RushDaemonHost { const paths: IDaemonPaths = resolveDaemonPathsFromProcess(workspaceKey); const sessions: Set = new Set(); const lifecycle: { closing: boolean } = { closing: false }; + const workspaceSessionProvider: WorkspaceSessionProvider = new WorkspaceSessionProvider( + options.createWorkspaceSessionAsync ?? WorkspaceSession.createAsync, + { + onError: options.onError, + repoRoot: canonicalRepoRoot, + rushVersion: options.rushVersion + } + ); const startedAtMs: number = Date.now(); - const listener: DaemonFrameListener = await DaemonFrameListener.listenAsync(paths, { - protocolVersion: DAEMON_PROTOCOL_VERSION, - startedAt: new Date(startedAtMs).toISOString(), - onConnection: (connection: DaemonFrameConnection) => { - const session: DaemonControlSession = new DaemonControlSession(connection, { - daemonVersion: options.daemonVersion, - startedAtMs, - onClosed: (closedSession: DaemonControlSession, error: Error | undefined) => { - sessions.delete(closedSession); - if (error) { - options.onError?.(error); - } - }, - onError: (error: Error) => options.onError?.(error) - }); - sessions.add(session); - if (lifecycle.closing) { - void session.closeAsync(); + await workspaceSessionProvider.getSessionAsync(); + let listener: DaemonFrameListener; + try { + listener = await DaemonFrameListener.listenAsync(paths, { + protocolVersion: DAEMON_PROTOCOL_VERSION, + startedAt: new Date(startedAtMs).toISOString(), + onConnection: (connection: DaemonFrameConnection) => { + const session: DaemonControlSession = new DaemonControlSession(connection, { + daemonVersion: options.daemonVersion, + startedAtMs, + onClosed: (closedSession: DaemonControlSession, error: Error | undefined) => { + sessions.delete(closedSession); + if (error) { + options.onError?.(error); + } + }, + onError: (error: Error) => options.onError?.(error) + }); + sessions.add(session); + if (lifecycle.closing) { + void session.closeAsync(); + } } + }); + } catch (error) { + try { + await workspaceSessionProvider[Symbol.asyncDispose](); + } catch (cleanupError) { + throw new AggregateError( + [error, cleanupError], + 'Failed to bind the daemon listener and dispose its workspace session.' + ); } - }); - return new RushDaemonHost(listener, paths, sessions, lifecycle); + throw error; + } + return new RushDaemonHost( + listener, + paths, + sessions, + lifecycle, + workspaceSessionProvider + ); + } + + /** Returns the single warm workspace session owned by this host. */ + public getWorkspaceSessionAsync(): Promise { + return this._workspaceSessionProvider.getSessionAsync(); } /** Closes active connections, stops listening, and removes transport artifacts. */ @@ -102,7 +143,29 @@ export class RushDaemonHost { private async _closeOnceAsync(): Promise { this._lifecycle.closing = true; - await Promise.all(Array.from(this._sessions, (session: DaemonControlSession) => session.closeAsync())); - await this._listener.closeAsync(); + const errors: unknown[] = []; + try { + await Promise.all( + Array.from(this._sessions, (session: DaemonControlSession) => session.closeAsync()) + ); + } catch (error) { + errors.push(error); + } + try { + await this._listener.closeAsync(); + } catch (error) { + errors.push(error); + } + try { + await this._workspaceSessionProvider[Symbol.asyncDispose](); + } catch (error) { + errors.push(error); + } + + if (errors.length === 1) { + throw errors[0]; + } else if (errors.length > 1) { + throw new AggregateError(errors, 'Failed to close Rush daemon host resources.'); + } } } diff --git a/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts new file mode 100644 index 0000000000..ee67d5afa1 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceInvalidationTracker.ts @@ -0,0 +1,98 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** + * A point-in-time view of workspace changes that have not yet been reconciled. + * + * @beta + */ +export interface IWorkspaceInvalidationSnapshot { + /** Paths reported by the watcher, sorted for deterministic consumption. */ + readonly changedPaths: ReadonlyArray; + /** True when the watcher reported a change without a path or encountered a watcher error. */ + readonly hasUnknownChanges: boolean; + /** False after a watcher error makes subsequent change detection unreliable. */ + readonly isWatcherHealthy: boolean; + /** The latest invalidation sequence included in this snapshot. */ + readonly sequence: number; +} + +const MAX_TRACKED_CHANGED_PATHS: number = 10_000; + +/** + * Retains workspace invalidations until a future request explicitly acknowledges them. + * + * @beta + */ +export class WorkspaceInvalidationTracker { + readonly #sequenceByPath: Map = new Map(); + #latestSequence: number = 0; + #unknownChangeSequence: number | undefined; + #watcherHealthy: boolean = true; + + /** Records a path-specific or unknown workspace change. */ + public invalidate(changedPath?: string): void { + const sequence: number = ++this.#latestSequence; + if (changedPath === undefined || this.#unknownChangeSequence !== undefined) { + this.#unknownChangeSequence = sequence; + return; + } + + if ( + !this.#sequenceByPath.has(changedPath) && + this.#sequenceByPath.size >= MAX_TRACKED_CHANGED_PATHS + ) { + this.#sequenceByPath.clear(); + this.#unknownChangeSequence = sequence; + return; + } + + this.#sequenceByPath.set(changedPath, sequence); + } + + /** + * Permanently marks the current watcher as unhealthy. + * + * Unknown invalidation remains pending so consumers cannot mistake the workspace for clean. + */ + public markWatcherUnhealthy(): void { + if (this.#watcherHealthy) { + this.#watcherHealthy = false; + this.invalidate(); + } + } + + /** Returns all changes that have not been acknowledged. */ + public getSnapshot(): IWorkspaceInvalidationSnapshot { + return { + changedPaths: Array.from(this.#sequenceByPath.keys()).sort(), + hasUnknownChanges: this.#unknownChangeSequence !== undefined, + isWatcherHealthy: this.#watcherHealthy, + sequence: this.#latestSequence + }; + } + + /** + * Acknowledges changes through a previously observed sequence. + * + * Changes that arrive after that sequence remain pending, including repeated changes to the same path. + */ + public acknowledgeThrough(sequence: number): void { + if (!Number.isSafeInteger(sequence) || sequence < 0 || sequence > this.#latestSequence) { + throw new RangeError(`Invalid workspace invalidation sequence: ${sequence}`); + } + + for (const [changedPath, pathSequence] of this.#sequenceByPath) { + if (pathSequence <= sequence) { + this.#sequenceByPath.delete(changedPath); + } + } + if ( + this.#watcherHealthy && + this.#unknownChangeSequence !== undefined && + this.#unknownChangeSequence <= sequence + ) { + this.#unknownChangeSequence = undefined; + } + } +} diff --git a/libraries/rush-daemon/src/WorkspaceSession.ts b/libraries/rush-daemon/src/WorkspaceSession.ts new file mode 100644 index 0000000000..8abe498e1f --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSession.ts @@ -0,0 +1,267 @@ +// 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 extends AsyncDisposable { + startAsync(onInvalidation: (changedPath?: string) => void): Promise; +} + +/** + * Optional engine state supplied when reusable graph construction is available. + * + * @remarks + * The default session intentionally leaves the graph, plugin session, and inputs snapshot uninitialized. + * Their existing construction is command-specific and remains blocked on the reusable runner lifetime work. + * + * @beta + */ +export interface IWorkspaceSessionComponents extends AsyncDisposable { + readonly 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; +} + +/** + * Context for constructing optional reusable workspace engine components. + * + * @beta + */ +export interface ICreateWorkspaceSessionComponentsOptions { + readonly invalidations: WorkspaceInvalidationTracker; + readonly onError?: (error: Error) => void; + readonly rushConfiguration: RushConfiguration; +} + +/** + * Constructs optional reusable graph, plugin, snapshot, and watcher state. + * + * @beta + */ +export type CreateWorkspaceSessionComponentsAsync = ( + options: ICreateWorkspaceSessionComponentsOptions +) => Promise; + +/** + * Options for initializing a workspace session. + * + * @beta + */ +export interface IWorkspaceSessionOptions { + readonly createComponentsAsync?: CreateWorkspaceSessionComponentsAsync; + readonly onError?: (error: Error) => void; + readonly repoRoot: string; + readonly rushVersion: string; +} + +/** + * The reusable state owned by one daemon lifecycle. + * + * @beta + */ +export interface IWorkspaceSession extends AsyncDisposable { + readonly inputsSnapshot: IInputsSnapshot | undefined; + readonly invalidations: WorkspaceInvalidationTracker; + readonly metadata: IWorkspaceSessionMetadata; + readonly operationGraph: IOperationGraph | undefined; + readonly rushConfiguration: RushConfiguration; + readonly rushSession: RushSession | undefined; +} + +/** + * Factory used by the daemon host to initialize its workspace session. + * + * @beta + */ +export type WorkspaceSessionFactory = (options: IWorkspaceSessionOptions) => Promise; + +const EMPTY_WORKSPACE_SESSION_COMPONENTS: IWorkspaceSessionComponents = { + [Symbol.asyncDispose]: () => Promise.resolve() +}; + +/** + * A warm workspace session with client-independent invalidation tracking. + * + * @beta + */ +export class WorkspaceSession implements IWorkspaceSession { + readonly #components: IWorkspaceSessionComponents; + readonly #sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined; + #disposePromise: Promise | undefined; + + 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, + sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined + ) { + this.rushConfiguration = rushConfiguration; + this.metadata = metadata; + this.invalidations = invalidations; + this.#components = components; + this.#sessionOwnedProjectWatcher = sessionOwnedProjectWatcher; + this.inputsSnapshot = components.inputsSnapshot; + this.operationGraph = components.operationGraph; + this.rushSession = components.rushSession; + } + + /** Loads workspace identity, creates reusable components, and starts headless invalidation tracking. */ + public static async createAsync(options: IWorkspaceSessionOptions): Promise { + const rushConfiguration: RushConfiguration = RushConfiguration.loadFromConfigurationFile( + path.join(options.repoRoot, 'rush.json') + ); + const canonicalRepoRoot: string = path.resolve(options.repoRoot); + if (path.resolve(rushConfiguration.rushJsonFolder) !== canonicalRepoRoot) { + throw new Error(`Rush configuration resolved outside the daemon workspace: ${options.repoRoot}`); + } + + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + const components: IWorkspaceSessionComponents = + (await options.createComponentsAsync?.({ + invalidations, + onError: options.onError, + rushConfiguration + })) ?? EMPTY_WORKSPACE_SESSION_COMPONENTS; + let projectWatcher: IWorkspaceInvalidationWatcher | undefined = components.projectWatcher; + let sessionOwnedProjectWatcher: IWorkspaceInvalidationWatcher | undefined; + try { + const metadata: IWorkspaceSessionMetadata = createMetadata( + rushConfiguration, + options.rushVersion + ); + if (!projectWatcher) { + projectWatcher = new WorkspaceSessionFileWatcher({ + onError: (error: Error) => { + invalidations.markWatcherUnhealthy(); + options.onError?.(error); + }, + rushConfiguration + }); + sessionOwnedProjectWatcher = projectWatcher; + } + const session: WorkspaceSession = new WorkspaceSession( + rushConfiguration, + metadata, + invalidations, + components, + sessionOwnedProjectWatcher + ); + await projectWatcher.startAsync((changedPath: string | undefined) => + invalidations.invalidate(changedPath) + ); + // Changes before the watcher registered its callbacks cannot be observed path-by-path. + invalidations.invalidate(); + return session; + } catch (error) { + const cleanupErrors: unknown[] = []; + try { + await sessionOwnedProjectWatcher?.[Symbol.asyncDispose](); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + try { + await components[Symbol.asyncDispose](); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + if (cleanupErrors.length > 0) { + throw new AggregateError( + [error, ...cleanupErrors], + 'Failed to initialize and clean up the workspace session.' + ); + } + throw error; + } + } + + /** Stops invalidation tracking and disposes injected engine resources. */ + public [Symbol.asyncDispose](): Promise { + this.#disposePromise ??= this.#disposeOnceAsync(); + return this.#disposePromise; + } + + async #disposeOnceAsync(): Promise { + let watcherError: unknown; + try { + await this.#sessionOwnedProjectWatcher?.[Symbol.asyncDispose](); + } catch (error) { + watcherError = error; + } + + try { + await this.#components[Symbol.asyncDispose](); + } catch (componentError) { + if (watcherError !== undefined) { + throw new AggregateError( + [watcherError, componentError], + 'Failed to dispose workspace session resources.' + ); + } + throw componentError; + } + if (watcherError !== undefined) { + throw watcherError; + } + } +} + +function createMetadata( + rushConfiguration: RushConfiguration, + rushVersion: string +): IWorkspaceSessionMetadata { + const projectNames: string[] = Array.from( + rushConfiguration.projects, + (project) => project.packageName + ).sort(); + return { + projectCount: projectNames.length, + projectNames, + repoRoot: rushConfiguration.rushJsonFolder, + rushJsonFile: rushConfiguration.rushJsonFile, + rushVersion + }; +} diff --git a/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts new file mode 100644 index 0000000000..acee521b21 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSessionFileWatcher.ts @@ -0,0 +1,121 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import { once } from 'node:events'; + +import type { RushConfiguration } from '@microsoft/rush-lib'; + +import type { IWorkspaceInvalidationWatcher } from './WorkspaceSession'; + +export interface IWorkspaceSessionFileWatcherOptions { + readonly onError?: (error: Error) => void; + readonly rushConfiguration: RushConfiguration; + readonly watchFactory?: WorkspaceWatchFactory; +} + +interface IWatchPath { + readonly folderPath: string; + readonly recursive: boolean; +} + +type WorkspaceWatchFactory = ( + folderPath: string, + options: { encoding: 'utf8'; recursive: boolean }, + listener: fs.WatchListener +) => fs.FSWatcher; + +const PATH_SEGMENT_SEPARATOR_REGEXP: RegExp = /[\\/]/; + +export class WorkspaceSessionFileWatcher implements IWorkspaceInvalidationWatcher { + readonly #onError: ((error: Error) => void) | undefined; + readonly #watchFactory: WorkspaceWatchFactory; + readonly #watchPaths: ReadonlyArray; + readonly #watchers: Set = new Set(); + #onInvalidation: ((changedPath?: string) => void) | undefined; + #disposed: boolean = false; + + public constructor(options: IWorkspaceSessionFileWatcherOptions) { + this.#onError = options.onError; + this.#watchFactory = options.watchFactory ?? fs.watch; + this.#watchPaths = getWatchPaths(options.rushConfiguration); + } + + public async startAsync(onInvalidation: (changedPath?: string) => void): Promise { + if (this.#disposed) { + throw new Error('The workspace watcher has already been disposed.'); + } + if (this.#onInvalidation) { + throw new Error('The workspace watcher has already been started.'); + } + + this.#onInvalidation = onInvalidation; + for (const watchPath of this.#watchPaths) { + this.#watchers.add(this.#createWatcher(watchPath)); + } + } + + public async [Symbol.asyncDispose](): Promise { + if (this.#disposed) { + return; + } + this.#disposed = true; + const closePromises: Promise[] = []; + for (const watcher of this.#watchers) { + closePromises.push(once(watcher, 'close')); + watcher.close(); + } + await Promise.all(closePromises); + this.#watchers.clear(); + this.#onInvalidation = undefined; + } + + #createWatcher(watchPath: IWatchPath): fs.FSWatcher { + const watcher: fs.FSWatcher = this.#watchFactory( + watchPath.folderPath, + { encoding: 'utf8', recursive: watchPath.recursive }, + (eventType: string, filename: string | null) => { + void eventType; + const changedFilename: string | undefined = filename ?? undefined; + if (!isIgnoredPath(changedFilename)) { + this.#onInvalidation?.( + changedFilename === undefined + ? undefined + : path.resolve(watchPath.folderPath, changedFilename) + ); + } + } + ); + watcher.on('error', (error: Error) => { + this.#onInvalidation?.(); + this.#onError?.(error); + }); + watcher.once('close', () => this.#watchers.delete(watcher)); + watcher.unref(); + return watcher; + } +} + +function getWatchPaths(rushConfiguration: RushConfiguration): ReadonlyArray { + const recursiveFolders: Set = new Set([rushConfiguration.commonRushConfigFolder]); + for (const subspace of rushConfiguration.subspaces) { + recursiveFolders.add(subspace.getSubspaceConfigFolderPath()); + } + for (const project of rushConfiguration.projects) { + recursiveFolders.add(project.projectFolder); + } + return [ + { folderPath: rushConfiguration.rushJsonFolder, recursive: false }, + ...Array.from(recursiveFolders, (folderPath: string) => ({ folderPath, recursive: true })) + ]; +} + +function isIgnoredPath(filename: string | undefined): boolean { + if (filename === undefined) { + return false; + } + return filename + .split(PATH_SEGMENT_SEPARATOR_REGEXP) + .some((segment: string) => segment === '.git' || segment === 'node_modules'); +} diff --git a/libraries/rush-daemon/src/WorkspaceSessionProvider.ts b/libraries/rush-daemon/src/WorkspaceSessionProvider.ts new file mode 100644 index 0000000000..85ba8eba51 --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceSessionProvider.ts @@ -0,0 +1,83 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IWorkspaceSession, + IWorkspaceSessionOptions, + WorkspaceSessionFactory +} from './WorkspaceSession'; + +export class WorkspaceSessionProvider implements AsyncDisposable { + readonly #factory: WorkspaceSessionFactory; + readonly #options: IWorkspaceSessionOptions; + #disposePromise: Promise | undefined; + #initializationDisposalPromise: Promise | undefined; + #initializationPromise: Promise | undefined; + #session: IWorkspaceSession | undefined; + #disposed: boolean = false; + + public constructor(factory: WorkspaceSessionFactory, options: IWorkspaceSessionOptions) { + this.#factory = factory; + this.#options = options; + } + + public getSessionAsync(): Promise { + if (this.#disposed) { + return Promise.reject(new Error('The workspace session provider has been disposed.')); + } + if (this.#session) { + return Promise.resolve(this.#session); + } + if (!this.#initializationPromise) { + const initializationPromise: Promise = Promise.resolve().then(() => + this.#initializeAsync() + ); + this.#initializationPromise = initializationPromise; + void initializationPromise.catch(() => { + if (this.#initializationPromise === initializationPromise) { + this.#initializationPromise = undefined; + } + }); + } + return this.#initializationPromise; + } + + public [Symbol.asyncDispose](): Promise { + this.#disposePromise ??= this.#disposeOnceAsync(); + return this.#disposePromise; + } + + async #disposeOnceAsync(): Promise { + this.#disposed = true; + try { + const session: IWorkspaceSession | undefined = + this.#session ?? + (await this.#initializationPromise?.then( + (initializedSession: IWorkspaceSession) => initializedSession, + () => undefined + )); + if (session) { + await session[Symbol.asyncDispose](); + } else { + await this.#initializationDisposalPromise; + } + } finally { + this.#session = undefined; + this.#initializationPromise = undefined; + this.#initializationDisposalPromise = undefined; + } + } + + async #initializeAsync(): Promise { + const session: IWorkspaceSession = await this.#factory(this.#options); + if (this.#disposed) { + this.#initializationDisposalPromise = Promise.resolve().then(() => + session[Symbol.asyncDispose]() + ); + await this.#initializationDisposalPromise; + throw new Error('The workspace session provider was disposed during initialization.'); + } + this.#session = session; + return session; + } +} diff --git a/libraries/rush-daemon/src/index.ts b/libraries/rush-daemon/src/index.ts index a9178e9749..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..020d7e5ed3 100644 --- a/libraries/rush-daemon/src/test/RushDaemonHost.test.ts +++ b/libraries/rush-daemon/src/test/RushDaemonHost.test.ts @@ -17,8 +17,10 @@ import type { IDaemonFrame } from '@rushstack/rush-daemon-protocol'; import { + computeDaemonWorkspaceKey, connectDaemonAsync, - readDaemonLockfile + readDaemonLockfile, + resolveDaemonPathsFromProcess } from '@rushstack/rush-daemon-transport'; import type { DaemonFrameConnection, @@ -26,7 +28,10 @@ import type { } from '@rushstack/rush-daemon-transport'; import { RushDaemonHost } from '../RushDaemonHost'; +import type { IRushDaemonHostOptions } from '../RushDaemonHost'; import { serveRushDaemonAsync } from '../serveRushDaemon'; +import type { IWorkspaceSession } from '../WorkspaceSession'; +import { TestWorkspaceSession } from './TestWorkspaceSession'; const RUSH_VERSION: string = '5.178.1'; const DAEMON_VERSION: string = '0.1.0-test'; @@ -48,12 +53,17 @@ function createTestRepoRoot(): string { return repoRoot; } -function createHostOptions(repoRoot: string): { - daemonVersion: string; - repoRoot: string; - rushVersion: string; -} { - return { daemonVersion: DAEMON_VERSION, repoRoot, rushVersion: RUSH_VERSION }; +function createHostOptions( + repoRoot: string, + overrides: Partial = {} +): IRushDaemonHostOptions { + return { + daemonVersion: DAEMON_VERSION, + repoRoot, + rushVersion: RUSH_VERSION, + createWorkspaceSessionAsync: () => Promise.resolve(new TestWorkspaceSession(repoRoot)), + ...overrides + }; } async function exchangeControlAsync( @@ -134,16 +144,99 @@ describe(RushDaemonHost.name, () => { }); it('closes active connections and removes transport artifacts', async () => { + const disposalEvents: string[] = []; + const repoRoot: string = createTestRepoRoot(); const host: RushDaemonHost = await RushDaemonHost.startAsync( - createHostOptions(createTestRepoRoot()) + createHostOptions(repoRoot, { + createWorkspaceSessionAsync: () => + Promise.resolve( + new TestWorkspaceSession(repoRoot, () => disposalEvents.push('workspace-session')) + ) + }) ); const client: DaemonFrameConnection = await connectDaemonAsync(host.paths.socketPath); - const closed: Promise = new Promise((resolve: () => void) => client.onClosed(() => resolve())); + const closed: Promise = new Promise((resolve: () => void) => + client.onClosed(() => { + disposalEvents.push('client'); + resolve(); + }) + ); await host.closeAsync(); await closed; + expect(disposalEvents).toHaveLength(2); + expect(disposalEvents).toContain('client'); + expect(disposalEvents).toContain('workspace-session'); expect(readDaemonLockfile(host.paths.lockfilePath)).toBeUndefined(); await expect(connectDaemonAsync(host.paths.socketPath)).rejects.toMatchObject({ code: 'connectionRefused' }); }); + + it('initializes one workspace session and reuses it', async () => { + const repoRoot: string = createTestRepoRoot(); + const workspaceSession: IWorkspaceSession = new TestWorkspaceSession(repoRoot); + let factoryCalls: number = 0; + const host: RushDaemonHost = await RushDaemonHost.startAsync( + createHostOptions(repoRoot, { + createWorkspaceSessionAsync: () => { + factoryCalls++; + return Promise.resolve(workspaceSession); + } + }) + ); + try { + const [first, second] = await Promise.all([ + host.getWorkspaceSessionAsync(), + host.getWorkspaceSessionAsync() + ]); + expect(first).toBe(workspaceSession); + expect(second).toBe(workspaceSession); + expect(factoryCalls).toBe(1); + } finally { + await host.closeAsync(); + } + }); + + it('removes transport artifacts when workspace initialization fails', async () => { + const repoRoot: string = createTestRepoRoot(); + const workspaceKey: string = computeDaemonWorkspaceKey({ + canonicalRepoRoot: fs.realpathSync(repoRoot), + rushVersion: RUSH_VERSION + }); + const paths: IDaemonPaths = resolveDaemonPathsFromProcess(workspaceKey); + + await expect( + RushDaemonHost.startAsync( + createHostOptions(repoRoot, { + createWorkspaceSessionAsync: () => Promise.reject(new Error('workspace failed')) + }) + ) + ).rejects.toThrow('workspace failed'); + + expect(readDaemonLockfile(paths.lockfilePath)).toBeUndefined(); + await expect(connectDaemonAsync(paths.socketPath)).rejects.toMatchObject({ + code: 'connectionRefused' + }); + }); + + it('disposes the workspace session when listener binding fails', async () => { + const repoRoot: string = createTestRepoRoot(); + const firstHost: RushDaemonHost = await RushDaemonHost.startAsync(createHostOptions(repoRoot)); + const disposalEvents: string[] = []; + try { + await expect( + RushDaemonHost.startAsync( + createHostOptions(repoRoot, { + createWorkspaceSessionAsync: () => + Promise.resolve( + new TestWorkspaceSession(repoRoot, () => disposalEvents.push('workspace-session')) + ) + }) + ) + ).rejects.toMatchObject({ code: 'daemonAlreadyRunning' }); + expect(disposalEvents).toEqual(['workspace-session']); + } finally { + await firstHost.closeAsync(); + } + }); }); diff --git a/libraries/rush-daemon/src/test/TestWorkspaceSession.ts b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts new file mode 100644 index 0000000000..fb609e0c73 --- /dev/null +++ b/libraries/rush-daemon/src/test/TestWorkspaceSession.ts @@ -0,0 +1,48 @@ +// 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 { + readonly #onDispose: (() => unknown) | undefined; + + public readonly inputsSnapshot: IInputsSnapshot | undefined; + public readonly invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + public readonly metadata: IWorkspaceSessionMetadata; + public readonly operationGraph: IOperationGraph | undefined; + public readonly rushConfiguration: RushConfiguration = TEST_RUSH_CONFIGURATION; + public readonly rushSession: RushSession | undefined; + + public constructor(repoRoot: string, onDispose?: () => unknown) { + this.#onDispose = onDispose; + this.metadata = { + projectCount: 0, + projectNames: [], + repoRoot, + rushJsonFile: path.join(repoRoot, 'rush.json'), + rushVersion: '5.178.0' + }; + } + + public async [Symbol.asyncDispose](): Promise { + await this.#onDispose?.(); + } +} 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..5f22668fa6 --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceSession.test.ts @@ -0,0 +1,216 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IWorkspaceInvalidationWatcher, + IWorkspaceSessionComponents +} from '../WorkspaceSession'; +import { WorkspaceSession } from '../WorkspaceSession'; +import type { IWorkspaceInvalidationSnapshot } from '../WorkspaceInvalidationTracker'; +import { WorkspaceInvalidationTracker } from '../WorkspaceInvalidationTracker'; +import { TEST_REPO_ROOT } from './TestWorkspaceSession'; + +class TestInvalidationWatcher implements IWorkspaceInvalidationWatcher { + readonly #events: string[]; + #onInvalidation: ((changedPath?: string) => void) | undefined; + + public constructor(events: string[]) { + this.#events = events; + } + + public startAsync(onInvalidation: (changedPath?: string) => void): Promise { + this.#events.push('watcher-start'); + this.#onInvalidation = onInvalidation; + return Promise.resolve(); + } + + public invalidate(changedPath?: string): void { + if (!this.#onInvalidation) { + throw new Error('The test watcher is not running.'); + } + this.#onInvalidation(changedPath); + } + + public [Symbol.asyncDispose](): Promise { + this.#events.push('watcher-dispose'); + this.#onInvalidation = undefined; + return Promise.resolve(); + } +} + +describe(WorkspaceSession.name, () => { + it('loads stable metadata and retains headless invalidations until acknowledged', async () => { + const events: string[] = []; + const watcher: TestInvalidationWatcher = new TestInvalidationWatcher(events); + let componentFactoryCalls: number = 0; + const session: WorkspaceSession = await WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => { + componentFactoryCalls++; + return Promise.resolve({ + projectWatcher: watcher, + [Symbol.asyncDispose]: async () => { + await watcher[Symbol.asyncDispose](); + events.push('components-dispose'); + } + }); + } + }); + + expect(componentFactoryCalls).toBe(1); + expect(session.metadata).toMatchObject({ + projectCount: session.rushConfiguration.projects.length, + repoRoot: session.rushConfiguration.rushJsonFolder, + rushJsonFile: session.rushConfiguration.rushJsonFile, + rushVersion: '5.178.0' + }); + expect(session.metadata.projectNames).toEqual( + Array.from(session.rushConfiguration.projectsByName.keys()).sort() + ); + + const initialSnapshot: IWorkspaceInvalidationSnapshot = session.invalidations.getSnapshot(); + expect(initialSnapshot).toEqual({ + changedPaths: [], + hasUnknownChanges: true, + isWatcherHealthy: true, + sequence: 1 + }); + session.invalidations.acknowledgeThrough(initialSnapshot.sequence); + + watcher.invalidate('packages/a/src/index.ts'); + const firstSnapshot: IWorkspaceInvalidationSnapshot = session.invalidations.getSnapshot(); + watcher.invalidate('packages/a/src/index.ts'); + watcher.invalidate(); + session.invalidations.acknowledgeThrough(firstSnapshot.sequence); + + expect(session.invalidations.getSnapshot()).toEqual({ + changedPaths: ['packages/a/src/index.ts'], + hasUnknownChanges: true, + isWatcherHealthy: true, + sequence: 4 + }); + + await session[Symbol.asyncDispose](); + expect(events).toEqual(['watcher-start', 'watcher-dispose', 'components-dispose']); + await session[Symbol.asyncDispose](); + expect(events).toEqual(['watcher-start', 'watcher-dispose', 'components-dispose']); + }); + + it('does not allow a watcher error to be acknowledged as clean', async () => { + const watcher: TestInvalidationWatcher = new TestInvalidationWatcher([]); + const session: WorkspaceSession = await WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: watcher, + [Symbol.asyncDispose]: () => watcher[Symbol.asyncDispose]() + }) + }); + + session.invalidations.markWatcherUnhealthy(); + const snapshot: IWorkspaceInvalidationSnapshot = session.invalidations.getSnapshot(); + session.invalidations.acknowledgeThrough(snapshot.sequence); + + expect(session.invalidations.getSnapshot()).toMatchObject({ + hasUnknownChanges: true, + isWatcherHealthy: false + }); + await session[Symbol.asyncDispose](); + }); + + it('compacts excessive path changes into an unknown invalidation', () => { + const invalidations: WorkspaceInvalidationTracker = new WorkspaceInvalidationTracker(); + for (let index: number = 0; index <= 10_000; index++) { + invalidations.invalidate(`packages/project-${index}/lib/output.js`); + } + + const overflowSnapshot: IWorkspaceInvalidationSnapshot = invalidations.getSnapshot(); + expect(overflowSnapshot).toEqual({ + changedPaths: [], + hasUnknownChanges: true, + isWatcherHealthy: true, + sequence: 10_001 + }); + + invalidations.invalidate('packages/later-change/src/index.ts'); + invalidations.acknowledgeThrough(overflowSnapshot.sequence); + expect(invalidations.getSnapshot()).toMatchObject({ + changedPaths: [], + hasUnknownChanges: true, + sequence: 10_002 + }); + invalidations.acknowledgeThrough(10_002); + expect(invalidations.getSnapshot().hasUnknownChanges).toBe(false); + }); + + it('disposes components when watcher startup fails', async () => { + const events: string[] = []; + const watcher: IWorkspaceInvalidationWatcher = { + startAsync: () => Promise.reject(new Error('watcher startup failed')), + [Symbol.asyncDispose]: () => { + events.push('watcher-dispose'); + return Promise.resolve(); + } + }; + + await expect( + WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: watcher, + [Symbol.asyncDispose]: async () => { + await watcher[Symbol.asyncDispose](); + events.push('components-dispose'); + } + }) + }) + ).rejects.toThrow('watcher startup failed'); + expect(events).toEqual(['watcher-dispose', 'components-dispose']); + }); + + it('preserves initialization and cleanup failures when watcher startup fails', async () => { + const watcher: IWorkspaceInvalidationWatcher = { + startAsync: () => Promise.reject(new Error('watcher startup failed')), + [Symbol.asyncDispose]: () => Promise.reject(new Error('watcher cleanup failed')) + }; + + let thrownError: unknown; + try { + await WorkspaceSession.createAsync({ + repoRoot: TEST_REPO_ROOT, + rushVersion: '5.178.0', + createComponentsAsync: () => + Promise.resolve({ + projectWatcher: watcher, + [Symbol.asyncDispose]: async () => { + let watcherError: unknown; + try { + await watcher[Symbol.asyncDispose](); + } catch (error) { + watcherError = error; + } + throw new AggregateError( + [watcherError, new Error('component cleanup failed')], + 'component bundle cleanup failed' + ); + } + }) + }); + } catch (error) { + thrownError = error; + } + + expect(thrownError).toBeInstanceOf(AggregateError); + const errors: unknown[] = (thrownError as AggregateError).errors; + expect(errors[0]).toEqual(new Error('watcher startup failed')); + expect(errors[1]).toBeInstanceOf(AggregateError); + expect((errors[1] as AggregateError).errors).toEqual([ + new Error('watcher cleanup failed'), + new Error('component cleanup failed') + ]); + }); +}); diff --git a/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts b/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts new file mode 100644 index 0000000000..1d8a364dc8 --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceSessionFileWatcher.test.ts @@ -0,0 +1,47 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type * as fs from 'node:fs'; +import { EventEmitter } from 'node:events'; + +import { WorkspaceSessionFileWatcher } from '../WorkspaceSessionFileWatcher'; +import { TEST_RUSH_CONFIGURATION } from './TestWorkspaceSession'; + +class TestFsWatcher extends EventEmitter { + public close(): void { + this.emit('close'); + } + + public ref(): this { + return this; + } + + public unref(): this { + return this; + } +} + +describe(WorkspaceSessionFileWatcher.name, () => { + it('watches every configured subspace config folder', async () => { + const watchedPaths: string[] = []; + const watcher: WorkspaceSessionFileWatcher = new WorkspaceSessionFileWatcher({ + rushConfiguration: TEST_RUSH_CONFIGURATION, + watchFactory: (folderPath: string) => { + watchedPaths.push(folderPath); + return new TestFsWatcher() as fs.FSWatcher; + } + }); + + await watcher.startAsync(() => {}); + + const subspaceConfigFolders: string[] = TEST_RUSH_CONFIGURATION.subspaces.map((subspace) => + subspace.getSubspaceConfigFolderPath() + ); + expect(subspaceConfigFolders.length).toBeGreaterThan(0); + expect(watchedPaths).toEqual( + expect.arrayContaining(subspaceConfigFolders) + ); + + await watcher[Symbol.asyncDispose](); + }); +}); diff --git a/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts new file mode 100644 index 0000000000..a5328edaac --- /dev/null +++ b/libraries/rush-daemon/src/test/WorkspaceSessionProvider.test.ts @@ -0,0 +1,148 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IWorkspaceSession, IWorkspaceSessionOptions } from '../WorkspaceSession'; +import { WorkspaceSessionProvider } from '../WorkspaceSessionProvider'; +import { TestWorkspaceSession } from './TestWorkspaceSession'; + +const OPTIONS: IWorkspaceSessionOptions = { + repoRoot: 'repo', + rushVersion: '5.178.0' +}; + +describe(WorkspaceSessionProvider.name, () => { + it('shares concurrent initialization and reuses the result', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot); + let resolveFactory: ((value: IWorkspaceSession) => void) | undefined; + let factoryCalls: number = 0; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider(() => { + factoryCalls++; + return new Promise((resolve) => { + resolveFactory = resolve; + }); + }, OPTIONS); + + const first: Promise = provider.getSessionAsync(); + const second: Promise = provider.getSessionAsync(); + expect(first).toBe(second); + await Promise.resolve(); + expect(factoryCalls).toBe(1); + + resolveFactory?.(session); + await expect(first).resolves.toBe(session); + await expect(provider.getSessionAsync()).resolves.toBe(session); + expect(factoryCalls).toBe(1); + await provider[Symbol.asyncDispose](); + }); + + it('clears a failed initialization so a later attempt can retry', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot); + let factoryCalls: number = 0; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider(() => { + factoryCalls++; + return factoryCalls === 1 + ? Promise.reject(new Error('initialization failed')) + : Promise.resolve(session); + }, OPTIONS); + + await expect(provider.getSessionAsync()).rejects.toThrow('initialization failed'); + await expect(provider.getSessionAsync()).resolves.toBe(session); + expect(factoryCalls).toBe(2); + await provider[Symbol.asyncDispose](); + }); + + it('clears a synchronously thrown initialization so a later attempt can retry', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot); + let factoryCalls: number = 0; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider(() => { + factoryCalls++; + if (factoryCalls === 1) { + throw new Error('synchronous initialization failed'); + } + return Promise.resolve(session); + }, OPTIONS); + + await expect(provider.getSessionAsync()).rejects.toThrow('synchronous initialization failed'); + await expect(provider.getSessionAsync()).resolves.toBe(session); + expect(factoryCalls).toBe(2); + await provider[Symbol.asyncDispose](); + }); + + it('disposes a session that finishes initializing during shutdown', async () => { + const disposalEvents: string[] = []; + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot, () => + disposalEvents.push('session') + ); + let resolveFactory: ((value: IWorkspaceSession) => void) | undefined; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider( + () => + new Promise((resolve) => { + resolveFactory = resolve; + }), + OPTIONS + ); + + const initialization: Promise = provider.getSessionAsync(); + const disposal: Promise = provider[Symbol.asyncDispose](); + await Promise.resolve(); + resolveFactory?.(session); + + await expect(initialization).rejects.toThrow('disposed during initialization'); + await disposal; + expect(disposalEvents).toEqual(['session']); + }); + + it('surfaces a disposal failure from a session that finishes initializing during shutdown', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot, () => + Promise.reject(new Error('session cleanup failed')) + ); + let resolveFactory: ((value: IWorkspaceSession) => void) | undefined; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider( + () => + new Promise((resolve) => { + resolveFactory = resolve; + }), + OPTIONS + ); + + const initialization: Promise = provider.getSessionAsync(); + const disposal: Promise = provider[Symbol.asyncDispose](); + const initializationExpectation: Promise = expect(initialization).rejects.toThrow( + 'session cleanup failed' + ); + const disposalExpectation: Promise = expect(disposal).rejects.toThrow( + 'session cleanup failed' + ); + await Promise.resolve(); + resolveFactory?.(session); + + await Promise.all([initializationExpectation, disposalExpectation]); + }); + + it('surfaces a synchronous disposal failure during initialization shutdown', async () => { + const session: IWorkspaceSession = new TestWorkspaceSession(OPTIONS.repoRoot, () => { + throw new Error('synchronous session cleanup failed'); + }); + let resolveFactory: ((value: IWorkspaceSession) => void) | undefined; + const provider: WorkspaceSessionProvider = new WorkspaceSessionProvider( + () => + new Promise((resolve) => { + resolveFactory = resolve; + }), + OPTIONS + ); + + const initialization: Promise = provider.getSessionAsync(); + const disposal: Promise = provider[Symbol.asyncDispose](); + const initializationExpectation: Promise = expect(initialization).rejects.toThrow( + 'synchronous session cleanup failed' + ); + const disposalExpectation: Promise = expect(disposal).rejects.toThrow( + 'synchronous session cleanup failed' + ); + await Promise.resolve(); + resolveFactory?.(session); + + await Promise.all([initializationExpectation, disposalExpectation]); + }); +});