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..336d44ae1a --- /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": "patch" + } + ] +} diff --git a/common/reviews/api/rush-lib.api.md b/common/reviews/api/rush-lib.api.md index 6f21ecfaec..d99a8071cd 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; 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/IPCOperationRunner.ts b/libraries/rush-lib/src/logic/operations/IPCOperationRunner.ts index 6d8396591c..020e3732b0 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; } @@ -100,7 +96,6 @@ export class IPCOperationRunner implements IOperationRunner { const invalidate: (reason: string) => void = context.getInvalidateCallback(); return await context.runWithTerminalAsync( async (terminal: ITerminal, terminalProvider: ITerminalProvider): Promise => { - let isConnected: boolean = false; if (!this._ipcProcess || typeof this._ipcProcess.exitCode === 'number') { // Log any ignored parameters if (this._ignoredParameterValues.length > 0) { @@ -202,9 +197,7 @@ 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...'); const runCommand: IRunCommandMessage = { command: 'run' @@ -213,10 +206,6 @@ export class IPCOperationRunner implements IOperationRunner { }, reject); }); - if (isConnected && !this._persist) { - await this.closeAsync(); - } - // @rushstack/operation-graph does not currently have a concept of "Success with Warning" // To match existing ShellOperationRunner behavior we treat any stderr as a warning. return status === OperationStatus.Success && hasWarningOrError 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..ffaa73d802 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 @@ -448,11 +453,18 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera // Delegate global state reporting await executeContext.onResultAsync(this); } finally { - if (this.isTerminal) { - this._collatedWriter?.close(); - this.stdioSummarizer.close(); - this.problemCollector.close(); - } + this.finalize(); + } + } + + /** + * Closes per-record output resources after the record reaches a terminal state. + */ + public finalize(): void { + if (this.isTerminal) { + this._collatedWriter?.close(); + this.stdioSummarizer.close(); + this.problemCollector.close(); } } } diff --git a/libraries/rush-lib/src/logic/operations/OperationGraph.ts b/libraries/rush-lib/src/logic/operations/OperationGraph.ts index 8d01d74618..e56c113318 100644 --- a/libraries/rush-lib/src/logic/operations/OperationGraph.ts +++ b/libraries/rush-lib/src/logic/operations/OperationGraph.ts @@ -799,6 +799,8 @@ export class OperationGraph implements IOperationGraph { onStartAsync: onOperationStartAsync, onResultAsync: onOperationCompleteAsync }; + const recordsWithRunnerCleanup: Set = new Set(); + const graph: OperationGraph = this; if (!this.quietMode) { const plural: string = totalOperations === 1 ? '' : 's'; @@ -873,9 +875,36 @@ export class OperationGraph implements IOperationGraph { }); } + const recordsToClose: OperationExecutionRecord[] = []; + for (const record of executionRecords.values()) { + if (!recordsWithRunnerCleanup.has(record) && !record.shouldRunnerPersist) { + recordsToClose.push(record); + } + } + function reportRunnerCleanupFailure(record: OperationExecutionRecord, error: Error): void { + record.error = error; + record.status = OperationStatus.Failure; + _reportOperationErrorIfAny(record); + state.hasAnyFailures = true; + } + await Async.forEachAsync( + recordsToClose, + async (record: OperationExecutionRecord) => { + try { + await this.closeRunnersAsync([record.operation]); + } catch (e) { + reportRunnerCleanupFailure(record, e); + } + }, + { concurrency: this.parallelism } + ); + for (const record of executionRecords.values()) { + record.finalize(); + } + 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; @@ -1027,6 +1056,16 @@ export class OperationGraph implements IOperationGraph { record.error = e; record.status = OperationStatus.Failure; } + if (!record.shouldRunnerPersist) { + recordsWithRunnerCleanup.add(record); + try { + await graph.closeRunnersAsync([record.operation]); + } catch (e) { + _reportOperationErrorIfAny(record); + record.error = e; + record.status = OperationStatus.Failure; + } + } _onOperationComplete(record, state); } } 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..7b21ab9128 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,30 @@ 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; + return await super.executeAsync(context); + } +} + +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 +1158,227 @@ 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(1); + expect(changingRunner.closeAsync).not.toHaveBeenCalled(); + + persistentOperations = new Set([persistentOperation]); + await graph.executeAsync({}); + expect(persistentRunner.closeAsync).not.toHaveBeenCalled(); + expect(oneShotRunner.closeAsync).toHaveBeenCalledTimes(2); + expect(changingRunner.closeAsync).toHaveBeenCalledTimes(1); + }); + + 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 () => { + 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(1); + 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