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..39f0fced3f 100644 --- a/common/reviews/api/rush-lib.api.md +++ b/common/reviews/api/rush-lib.api.md @@ -653,6 +653,7 @@ export interface IOperationGraphContext extends ICreateOperationsContext { export interface IOperationGraphIterationOptions { // (undocumented) inputsSnapshot?: IInputsSnapshot; + shouldRunnerPersist?: (operation: Operation) => boolean; startTime?: number; } @@ -721,6 +722,7 @@ export interface IOperationRunnerContext { createLogFile: boolean; logFileSuffix?: string; }): Promise; + shouldRunnerPersist?: boolean; status: OperationStatus; stopwatch: IStopwatchResult; } diff --git a/libraries/rush-lib/src/logic/operations/IOperationGraph.ts b/libraries/rush-lib/src/logic/operations/IOperationGraph.ts index 294a620567..768fb2c2bc 100644 --- a/libraries/rush-lib/src/logic/operations/IOperationGraph.ts +++ b/libraries/rush-lib/src/logic/operations/IOperationGraph.ts @@ -21,6 +21,16 @@ export interface IOperationGraphIterationOptions { * The time when the iteration was scheduled, if available, as returned by `performance.now()`. */ startTime?: number; + + /** + * Returns whether the operation's runner should remain active after this iteration. + * + * @remarks + * When omitted, all runners remain active. Returning `false` causes a runner that supports immediate + * teardown to close when its operation completes. Any cold runner that remains active because its operation + * did not execute is closed after the iteration. + */ + shouldRunnerPersist?: (operation: Operation) => boolean; } /** diff --git a/libraries/rush-lib/src/logic/operations/IOperationRunner.ts b/libraries/rush-lib/src/logic/operations/IOperationRunner.ts index e8dc3f2d10..5d4b27822d 100644 --- a/libraries/rush-lib/src/logic/operations/IOperationRunner.ts +++ b/libraries/rush-lib/src/logic/operations/IOperationRunner.ts @@ -69,6 +69,15 @@ export interface IOperationRunnerContext { */ error?: Error; + /** + * Whether the host wants this operation's runner to remain resident (kept warm) after the + * operation completes for the current iteration. Defaults to `true` when the host provides no + * persistence policy. Runners that own a long-lived child process (such as `IPCOperationRunner`) + * must release that process as soon as the operation completes when this is `false`, so that the + * subprocess does not consume memory while downstream operations execute. + */ + shouldRunnerPersist?: boolean; + /** * Returns a callback that invalidates this operation so that it will be re-executed in the next iteration. * The returned callback captures only the minimal state needed, avoiding retention of the full context. @@ -134,6 +143,8 @@ export interface IOperationRunner { * If true, this runner currently owns some kind of active resource (e.g. a service or a watch process). * This can be used to determine if the operation is "in progress" even if it is not currently executing. * If the runner supports this property, it should update it as appropriate during execution. + * A runner that closes its resource during `executeAsync()` when `context.shouldRunnerPersist` is false + * must report `false` afterward so that the graph's fallback cleanup does not close it again. * The property is optional to avoid breaking existing implementations of IOperationRunner. */ readonly isActive?: boolean; diff --git a/libraries/rush-lib/src/logic/operations/IPCOperationRunner.ts b/libraries/rush-lib/src/logic/operations/IPCOperationRunner.ts index 6d8396591c..21c3acb9e8 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; } @@ -213,7 +209,11 @@ export class IPCOperationRunner implements IOperationRunner { }, reject); }); - if (isConnected && !this._persist) { + // If the host does not want this runner kept warm for the next iteration, release the + // long-lived subprocess immediately now that the operation has completed. Deferring this to + // the end of the iteration would keep the subprocess resident (consuming memory) while + // downstream operations execute, risking RAM exhaustion. + if (isConnected && context.shouldRunnerPersist === false) { 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..7252795b80 100644 --- a/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts +++ b/libraries/rush-lib/src/logic/operations/OperationExecutionRecord.ts @@ -46,6 +46,7 @@ export interface IOperationExecutionRecordContext { onOperationStateChanged?: (record: OperationExecutionRecord) => void; createEnvironment?: (record: OperationExecutionRecord) => IEnvironment; invalidate?: (operations: Iterable, reason: string) => void; + shouldRunnerPersist?: (operation: Operation) => boolean; inputsSnapshot: IInputsSnapshot | undefined; maxParallelism: number; @@ -222,6 +223,14 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera return this._context.createEnvironment?.(this); } + /** + * Whether this operation's runner should remain resident after the operation completes for the + * current iteration. Defaults to `true` (kept warm) when the host supplies no persistence policy. + */ + public get shouldRunnerPersist(): boolean { + return this._context.shouldRunnerPersist?.(this.operation) ?? true; + } + public getInvalidateCallback(): (reason: string) => void { const invalidateFn: ((operations: Iterable, reason: string) => void) | undefined = this._context.invalidate; diff --git a/libraries/rush-lib/src/logic/operations/OperationGraph.ts b/libraries/rush-lib/src/logic/operations/OperationGraph.ts index 8d01d74618..5cbfc254c4 100644 --- a/libraries/rush-lib/src/logic/operations/OperationGraph.ts +++ b/libraries/rush-lib/src/logic/operations/OperationGraph.ts @@ -85,6 +85,7 @@ interface IExecutionIterationContext extends IOperationExecutionRecordContext { promise: Promise | undefined; startTime?: number; + shouldRunnerPersist: IOperationGraphIterationOptions['shouldRunnerPersist']; completedOperations: number; totalOperations: number; @@ -572,9 +573,16 @@ export class OperationGraph implements IOperationGraph { ): Promise { const { _getInputsSnapshotAsync: getInputsSnapshotAsync } = this; - const { startTime = performance.now(), inputsSnapshot = await getInputsSnapshotAsync?.() } = - iterationOptions; - const iterationOptionsForCallbacks: IOperationGraphIterationOptions = { startTime, inputsSnapshot }; + const { + startTime = performance.now(), + inputsSnapshot = await getInputsSnapshotAsync?.(), + shouldRunnerPersist + } = iterationOptions; + const iterationOptionsForCallbacks: IOperationGraphIterationOptions = { + startTime, + inputsSnapshot, + shouldRunnerPersist + }; const { hooks } = this; @@ -607,6 +615,7 @@ export class OperationGraph implements IOperationGraph { const iterationContext: IExecutionIterationContext = { abortController, startTime, + shouldRunnerPersist, streamCollator, terminal, inputsSnapshot, @@ -757,7 +766,8 @@ export class OperationGraph implements IOperationGraph { const iterationOptions: IOperationGraphIterationOptions = { inputsSnapshot: iterationContext.inputsSnapshot, - startTime: iterationContext.startTime + startTime: iterationContext.startTime, + shouldRunnerPersist: iterationContext.shouldRunnerPersist }; const executionQueue: AsyncOperationQueue = new AsyncOperationQueue( @@ -1011,6 +1021,20 @@ export class OperationGraph implements IOperationGraph { telemetry.log(logEntry); } + // IPC operations that executed this iteration release their runner immediately upon completion. + // This fallback closes cold runners that remain active because their operation did not execute, + // for example due to a cache hit, abort, blocked dependency, or disabled operation. + const { shouldRunnerPersist } = iterationContext; + if (shouldRunnerPersist) { + const operationsToClose: Operation[] = []; + for (const operation of this.operations) { + if (!shouldRunnerPersist(operation) && operation.runner?.isActive !== false) { + operationsToClose.push(operation); + } + } + await this.closeRunnersAsync(operationsToClose); + } + return status; // This function is a callback because it may write to the collatedWriter before 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..50ee72ec07 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,26 @@ 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); + // Mirror IPCOperationRunner: a runner that owns a long-lived resource must release it + // immediately upon completion when the host has opted this operation out of persistence for + // the current iteration, rather than waiting until the end of the iteration. + if (context.shouldRunnerPersist === false) { + await this.closeAsync(); + } + return status; + } +} + describe('OperationGraph', () => { let graphOptions: IOperationGraphOptions; let graphIterationOptions: IOperationGraphIterationOptions; @@ -1134,13 +1154,164 @@ 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 when no policy is provided', 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 per-operation 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 + ); + + await graph.executeAsync({ + shouldRunnerPersist: (operation) => operation !== oneShotOperation + }); + expect(persistentRunner.closeAsync).not.toHaveBeenCalled(); + expect(oneShotRunner.closeAsync).toHaveBeenCalledTimes(1); + expect(changingRunner.closeAsync).not.toHaveBeenCalled(); + + await graph.executeAsync({ + shouldRunnerPersist: (operation) => operation === persistentOperation + }); + 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 + ); + + await graph.executeAsync({ + shouldRunnerPersist: (operation) => operation !== upstreamOperation + }); + + 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'); + + 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); + + graph.hooks.beforeExecuteIterationAsync.tapPromise( + 'test', + async (): Promise => OperationStatus.FromCache + ); + + await graph.executeAsync({ + shouldRunnerPersist: (operation) => operation === persistentOperation + }); + + expect(coldRunner.closeAsync).toHaveBeenCalledTimes(1); + expect(coldRunner.isActive).toBe(false); + expect(persistentRunner.closeAsync).not.toHaveBeenCalled(); + expect(persistentRunner.isActive).toBe(true); + }); +}); + +describe('closeRunnersAsync', () => { it('invokes closeAsync on runners and triggers onExecutionStatesUpdated hook', async () => { const localOptions: IOperationGraphOptions = { quietMode: false,