diff --git a/common/changes/@microsoft/rush/bmiddha-runner-persist-policy.json b/common/changes/@microsoft/rush/bmiddha-runner-persist-policy.json new file mode 100644 index 0000000000..c084a6f109 --- /dev/null +++ b/common/changes/@microsoft/rush/bmiddha-runner-persist-policy.json @@ -0,0 +1,9 @@ +{ + "changes": [ + { + "packageName": "@microsoft/rush", + "comment": "Add a per-iteration, host-driven runner persist policy so IPC runners can be kept hot or torn down per operation per iteration, instead of fixing persistence at plugin construction.", + "type": "minor" + } + ] +} diff --git a/common/reviews/api/rush-lib.api.md b/common/reviews/api/rush-lib.api.md index 1583568c3a..835442a9e1 100644 --- a/common/reviews/api/rush-lib.api.md +++ b/common/reviews/api/rush-lib.api.md @@ -407,6 +407,7 @@ export interface ICobuildLockProvider { // @alpha export interface IConfigurableOperation extends IBaseOperationExecutionResult { enabled: boolean; + shouldRunnerPersist: boolean; } // @public @@ -613,6 +614,7 @@ export interface IOperationExecutionResult extends IBaseOperationExecutionResult readonly logFilePaths: ILogFilePaths | undefined; readonly nonCachedDurationMs: number | undefined; readonly problemCollector: IProblemCollector; + readonly shouldRunnerPersist: boolean; readonly silent: boolean; readonly status: OperationStatus; readonly stdioSummarizer: StdioSummarizer; @@ -721,6 +723,7 @@ export interface IOperationRunnerContext { createLogFile: boolean; logFileSuffix?: string; }): Promise; + readonly shouldRunnerPersist: boolean; status: OperationStatus; stopwatch: IStopwatchResult; } diff --git a/libraries/rush-lib/src/logic/operations/IOperationExecutionResult.ts b/libraries/rush-lib/src/logic/operations/IOperationExecutionResult.ts index 8df76ec1b6..dfe8e30715 100644 --- a/libraries/rush-lib/src/logic/operations/IOperationExecutionResult.ts +++ b/libraries/rush-lib/src/logic/operations/IOperationExecutionResult.ts @@ -67,6 +67,12 @@ export interface IConfigurableOperation extends IBaseOperationExecutionResult { * True if the operation should execute in this iteration, false otherwise. */ enabled: boolean; + + /** + * True if the operation's runner should remain active after this iteration, false otherwise. + * Defaults to true. + */ + shouldRunnerPersist: boolean; } /** @@ -94,6 +100,10 @@ export interface IOperationExecutionResult extends IBaseOperationExecutionResult * True if the operation should execute in this iteration, false otherwise. */ readonly enabled: boolean; + /** + * True if the operation's runner should remain active after this iteration, false otherwise. + */ + readonly shouldRunnerPersist: boolean; /** * Object tracking execution timing. */ diff --git a/libraries/rush-lib/src/logic/operations/IOperationRunner.ts b/libraries/rush-lib/src/logic/operations/IOperationRunner.ts index e8dc3f2d10..911f893f88 100644 --- a/libraries/rush-lib/src/logic/operations/IOperationRunner.ts +++ b/libraries/rush-lib/src/logic/operations/IOperationRunner.ts @@ -57,6 +57,11 @@ export interface IOperationRunnerContext { */ status: OperationStatus; + /** + * True if the runner should remain active after this execution, false otherwise. + */ + readonly shouldRunnerPersist: boolean; + /** * The environment in which the operation is being executed. * A return value of `undefined` indicates that it should inherit the environment from the parent process. diff --git a/libraries/rush-lib/src/logic/operations/IPCOperationRunner.ts b/libraries/rush-lib/src/logic/operations/IPCOperationRunner.ts index 6d8396591c..f6545fee09 100644 --- a/libraries/rush-lib/src/logic/operations/IPCOperationRunner.ts +++ b/libraries/rush-lib/src/logic/operations/IPCOperationRunner.ts @@ -28,7 +28,6 @@ export interface IIPCOperationRunnerOptions { initialCommand: string; incrementalCommand: string | undefined; commandForHash: string; - persist: boolean; ignoredParameterValues: ReadonlyArray; } @@ -58,7 +57,6 @@ export class IPCOperationRunner implements IOperationRunner { private readonly _initialCommand: string; private readonly _incrementalCommand: string | undefined; private readonly _commandForHash: string; - private readonly _persist: boolean; private readonly _ignoredParameterValues: ReadonlyArray; private _ipcProcess: ChildProcess | undefined; @@ -72,7 +70,6 @@ export class IPCOperationRunner implements IOperationRunner { initialCommand, incrementalCommand, commandForHash, - persist, ignoredParameterValues } = options; this.name = name; @@ -83,7 +80,6 @@ export class IPCOperationRunner implements IOperationRunner { this._incrementalCommand = incrementalCommand; this._commandForHash = commandForHash; - this._persist = persist; this._ignoredParameterValues = ignoredParameterValues; } @@ -202,7 +198,6 @@ export class IPCOperationRunner implements IOperationRunner { subProcess.on('message', finishHandler); subProcess.on('error', reject); subProcess.on('exit', onExit); - this._processReadyPromise!.then(() => { isConnected = true; terminal.writeLine('Child supports IPC protocol. Sending "run" command...'); @@ -213,7 +208,7 @@ export class IPCOperationRunner implements IOperationRunner { }, reject); }); - if (isConnected && !this._persist) { + if (isConnected && !context.shouldRunnerPersist) { await this.closeAsync(); } diff --git a/libraries/rush-lib/src/logic/operations/IPCOperationRunnerPlugin.ts b/libraries/rush-lib/src/logic/operations/IPCOperationRunnerPlugin.ts index 7d5cccd785..1823c22b4f 100644 --- a/libraries/rush-lib/src/logic/operations/IPCOperationRunnerPlugin.ts +++ b/libraries/rush-lib/src/logic/operations/IPCOperationRunnerPlugin.ts @@ -81,7 +81,6 @@ export class IPCOperationRunnerPlugin implements IPhasedCommandPlugin { initialCommand, incrementalCommand, commandForHash, - persist: true, ignoredParameterValues }); diff --git a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts index 2de7655d1d..424730af49 100644 --- a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts +++ b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts @@ -84,6 +84,11 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera */ public enabled: boolean; + /** + * If true, this operation's runner should remain active after this iteration. + */ + public shouldRunnerPersist: boolean = true; + /** * This number represents how far away this Operation is from the furthest "root" operation (i.e. * an operation with no consumers). This helps us to calculate the critical path (i.e. the diff --git a/libraries/rush-lib/src/logic/operations/OperationGraph.ts b/libraries/rush-lib/src/logic/operations/OperationGraph.ts index 8d01d74618..659e933457 100644 --- a/libraries/rush-lib/src/logic/operations/OperationGraph.ts +++ b/libraries/rush-lib/src/logic/operations/OperationGraph.ts @@ -390,7 +390,7 @@ export class OperationGraph implements IOperationGraph { } } - public async closeRunnersAsync(operations?: Operation[]): Promise { + public async closeRunnersAsync(operations?: Iterable): Promise { const promises: Promise[] = []; const recordMap: ReadonlyMap = this._currentIteration?.records ?? this.resultByOperation; @@ -873,9 +873,35 @@ export class OperationGraph implements IOperationGraph { }); } + const recordsToClose: OperationExecutionRecord[] = []; + for (const record of executionRecords.values()) { + if (!record.shouldRunnerPersist) { + recordsToClose.push(record); + } + } + function reportRunnerCleanupFailure(record: OperationExecutionRecord, error: Error): void { + record.error = error; + record.status = OperationStatus.Failure; + _reportOperationErrorIfAny(record); + state.hasAnyFailures = true; + } + if (recordsToClose.length > 0) { + try { + await this.closeRunnersAsync(recordsToClose.map((record) => record.operation)); + } catch (e) { + for (const record of recordsToClose) { + reportRunnerCleanupFailure(record, e); + } + } + } + for (const record of executionRecords.values()) { + record.stdioSummarizer.close(); + record.problemCollector.close(); + } + const status: OperationStatus = (() => { - if (bailStatus) return bailStatus; if (state.hasAnyFailures) return OperationStatus.Failure; + if (bailStatus) return bailStatus; if (state.hasAnyAborted) return OperationStatus.Aborted; if (state.hasAnyNonAllowedWarnings) return OperationStatus.SuccessWithWarning; if (iterationContext.totalOperations === 0) return OperationStatus.NoOp; diff --git a/libraries/rush-lib/src/logic/operations/test/OperationGraph.test.ts b/libraries/rush-lib/src/logic/operations/test/OperationGraph.test.ts index 913f15f705..8c9253c2a2 100644 --- a/libraries/rush-lib/src/logic/operations/test/OperationGraph.test.ts +++ b/libraries/rush-lib/src/logic/operations/test/OperationGraph.test.ts @@ -107,6 +107,34 @@ function createGraph( return new OperationGraph(new Set([operation]), graphOptions); } +class ClosableRunner extends MockOperationRunner { + public isActive: boolean = false; + + public readonly closeAsync: jest.Mock, []> = jest.fn(async () => { + this.isActive = false; + }); + + public override async executeAsync(context: IOperationRunnerContext): Promise { + this.isActive = true; + const status: OperationStatus = await super.executeAsync(context); + if (!context.shouldRunnerPersist) { + await this.closeAsync(); + } + return status; + } +} + +function configureRunnerPersistence( + graph: OperationGraph, + shouldRunnerPersist: (operation: Operation) => boolean +): void { + graph.hooks.configureIteration.tap('test-runner-persistence', (records) => { + for (const [operation, record] of records) { + record.shouldRunnerPersist = shouldRunnerPersist(operation); + } + }); +} + describe('OperationGraph', () => { let graphOptions: IOperationGraphOptions; let graphIterationOptions: IOperationGraphIterationOptions; @@ -1134,13 +1162,228 @@ describe('deferred invalidation during active iteration', () => { }); }); -describe('closeRunnersAsync', () => { - class ClosableRunner extends MockOperationRunner { - public readonly closeAsync: jest.Mock, []> = jest.fn(async () => { - /* no-op */ +describe('runner persistence policy', () => { + const graphOptions: IOperationGraphOptions = { + quietMode: false, + debugMode: false, + parallelism: 3, + allowOversubscription: true, + destinations: [mockWritable], + abortController: new AbortController() + }; + + it('keeps runners active across successive iterations by default', async () => { + const runAsync: jest.Mock, []> = jest.fn(async () => OperationStatus.Success); + const runner: ClosableRunner = new ClosableRunner('default-persistent', runAsync); + const graph: OperationGraph = createGraph(graphOptions, runner); + + await graph.executeAsync({}); + expect(runAsync).toHaveBeenCalledTimes(1); + expect(runner.closeAsync).not.toHaveBeenCalled(); + + await graph.executeAsync({}); + expect(runAsync).toHaveBeenCalledTimes(2); + expect(runner.closeAsync).not.toHaveBeenCalled(); + }); + + it('applies configured record policies independently on each iteration', async () => { + const persistentRunner: ClosableRunner = new ClosableRunner('persistent'); + const oneShotRunner: ClosableRunner = new ClosableRunner('one-shot'); + const changingRunner: ClosableRunner = new ClosableRunner('changing'); + + const persistentOperation: Operation = new Operation({ + runner: persistentRunner, + logFilenameIdentifier: 'persistent', + phase: mockPhase, + project: getOrCreateProject('persistent') + }); + const oneShotOperation: Operation = new Operation({ + runner: oneShotRunner, + logFilenameIdentifier: 'one-shot', + phase: mockPhase, + project: getOrCreateProject('one-shot') + }); + const changingOperation: Operation = new Operation({ + runner: changingRunner, + logFilenameIdentifier: 'changing', + phase: mockPhase, + project: getOrCreateProject('changing') + }); + + const graph: OperationGraph = new OperationGraph( + new Set([persistentOperation, oneShotOperation, changingOperation]), + graphOptions + ); + + let persistentOperations: ReadonlySet = new Set([persistentOperation, changingOperation]); + configureRunnerPersistence(graph, (operation) => persistentOperations.has(operation)); + + await graph.executeAsync({}); + expect(persistentRunner.closeAsync).not.toHaveBeenCalled(); + expect(oneShotRunner.closeAsync).toHaveBeenCalledTimes(2); + expect(changingRunner.closeAsync).not.toHaveBeenCalled(); + + persistentOperations = new Set([persistentOperation]); + await graph.executeAsync({}); + expect(persistentRunner.closeAsync).not.toHaveBeenCalled(); + expect(oneShotRunner.closeAsync).toHaveBeenCalledTimes(4); + expect(changingRunner.closeAsync).toHaveBeenCalledTimes(2); + }); + + it('releases a non-persistent runner before its dependents execute', async () => { + const executionOrder: string[] = []; + + const upstreamRunner: ClosableRunner = new ClosableRunner('upstream', async () => { + executionOrder.push('upstream:run'); + return OperationStatus.Success; + }); + upstreamRunner.closeAsync.mockImplementation(async () => { + if (upstreamRunner.isActive) { + executionOrder.push('upstream:close'); + } + upstreamRunner.isActive = false; + }); + + const downstreamRunner: ClosableRunner = new ClosableRunner('downstream', async () => { + executionOrder.push('downstream:run'); + return OperationStatus.Success; + }); + + const upstreamOperation: Operation = new Operation({ + runner: upstreamRunner, + logFilenameIdentifier: 'upstream', + phase: mockPhase, + project: getOrCreateProject('upstream') + }); + const downstreamOperation: Operation = new Operation({ + runner: downstreamRunner, + logFilenameIdentifier: 'downstream', + phase: mockPhase, + project: getOrCreateProject('downstream') + }); + downstreamOperation.addDependency(upstreamOperation); + + const graph: OperationGraph = new OperationGraph( + new Set([upstreamOperation, downstreamOperation]), + graphOptions + ); + + configureRunnerPersistence(graph, (operation) => operation !== upstreamOperation); + await graph.executeAsync({}); + + expect(upstreamRunner.closeAsync).toHaveBeenCalledTimes(2); + expect(downstreamRunner.closeAsync).not.toHaveBeenCalled(); + // The non-persistent upstream runner must be released immediately upon its own completion, + // before the downstream operation begins executing — not deferred to the end of the iteration. + expect(executionOrder).toEqual(['upstream:run', 'upstream:close', 'downstream:run']); + }); + + it('closes an active cold runner when an iteration bypasses runner execution', async () => { + const persistentRunner: ClosableRunner = new ClosableRunner('persistent'); + const coldRunner: ClosableRunner = new ClosableRunner('cold'); + let coldRunnerWasClosedBeforeAfterIteration: boolean = false; + + const persistentOperation: Operation = new Operation({ + runner: persistentRunner, + logFilenameIdentifier: 'persistent', + phase: mockPhase, + project: getOrCreateProject('persistent') + }); + const coldOperation: Operation = new Operation({ + runner: coldRunner, + logFilenameIdentifier: 'cold', + phase: mockPhase, + project: getOrCreateProject('cold') + }); + + const graph: OperationGraph = new OperationGraph( + new Set([persistentOperation, coldOperation]), + graphOptions + ); + + await graph.executeAsync({}); + expect(persistentRunner.isActive).toBe(true); + expect(coldRunner.isActive).toBe(true); + + configureRunnerPersistence(graph, (operation) => operation === persistentOperation); + graph.hooks.beforeExecuteIterationAsync.tapPromise( + 'test', + async (): Promise => OperationStatus.FromCache + ); + graph.hooks.afterExecuteIterationAsync.tapPromise('test', async (status) => { + coldRunnerWasClosedBeforeAfterIteration = coldRunner.isActive === false; + return status; }); - } + await graph.executeAsync({}); + + expect(coldRunner.closeAsync).toHaveBeenCalledTimes(1); + expect(coldRunner.isActive).toBe(false); + expect(coldRunnerWasClosedBeforeAfterIteration).toBe(true); + expect(persistentRunner.closeAsync).not.toHaveBeenCalled(); + expect(persistentRunner.isActive).toBe(true); + }); + + it('reports a bypassed runner cleanup failure before finalizing the iteration', async () => { + const persistentRunner: ClosableRunner = new ClosableRunner('persistent'); + const coldRunner: ClosableRunner = new ClosableRunner('cold'); + const closeError: Error = new Error('close failed'); + const lifecycleEvents: string[] = []; + + const persistentOperation: Operation = new Operation({ + runner: persistentRunner, + logFilenameIdentifier: 'persistent', + phase: mockPhase, + project: getOrCreateProject('persistent') + }); + const coldOperation: Operation = new Operation({ + runner: coldRunner, + logFilenameIdentifier: 'cold', + phase: mockPhase, + project: getOrCreateProject('cold') + }); + + const graph: OperationGraph = new OperationGraph( + new Set([persistentOperation, coldOperation]), + graphOptions + ); + + await graph.executeAsync({}); + + configureRunnerPersistence(graph, (operation) => operation === persistentOperation); + coldRunner.closeAsync.mockImplementationOnce(async () => { + lifecycleEvents.push('close'); + throw closeError; + }); + graph.hooks.beforeExecuteIterationAsync.tapPromise( + 'test', + async (): Promise => OperationStatus.FromCache + ); + let afterIterationStatus: OperationStatus | undefined; + graph.hooks.afterExecuteIterationAsync.tapPromise('test', async (status) => { + lifecycleEvents.push('after-iteration'); + afterIterationStatus = status; + return status; + }); + graph.hooks.afterExecuteIterationAsync.tap('test-summary', (status, records) => { + _printOperationStatus(mockTerminal, { status, operationResults: records }); + return status; + }); + + const result: IExecutionResult = await graph.executeAsync({}); + const coldResult: IOperationExecutionResult | undefined = result.operationResults.get(coldOperation); + + expect(lifecycleEvents).toEqual(['close', 'after-iteration']); + expect(afterIterationStatus).toBe(OperationStatus.Failure); + expect(result.status).toBe(OperationStatus.Failure); + expect(graph.status).toBe(OperationStatus.Failure); + expect(coldResult?.status).toBe(OperationStatus.Failure); + expect(coldResult?.error).toBe(closeError); + expect(persistentRunner.closeAsync).not.toHaveBeenCalled(); + }); +}); + +describe('closeRunnersAsync', () => { it('invokes closeAsync on runners and triggers onExecutionStatesUpdated hook', async () => { const localOptions: IOperationGraphOptions = { quietMode: false, diff --git a/libraries/rush-lib/src/pluginFramework/OperationGraphHooks.ts b/libraries/rush-lib/src/pluginFramework/OperationGraphHooks.ts index f72bf682de..e78552903d 100644 --- a/libraries/rush-lib/src/pluginFramework/OperationGraphHooks.ts +++ b/libraries/rush-lib/src/pluginFramework/OperationGraphHooks.ts @@ -42,7 +42,8 @@ export class OperationGraphHooks { /** * Hook invoked to decide what work a potential new iteration contains. * Use the `lastExecutedRecords` to determine which operations are new or have had their inputs changed. - * Set the `enabled` states on the values in `initialRecords` to control which operations will be executed. + * Set `enabled` and `shouldRunnerPersist` on the values in `initialRecords` to control which operations + * execute and which runners remain active after the iteration. * * @remarks * This hook is synchronous to guarantee that the `lastExecutedRecords` map remains stable for the