[rush] Add per-iteration runner persistence control

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"
+    }
+  ]
+}

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<T>;
+    readonly shouldRunnerPersist: boolean;
     status: OperationStatus;
     stopwatch: IStopwatchResult;
 }

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.
    */

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.

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<string>;
 }
 
@@ -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<string>;
 
   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();
         }
 

libraries/rush-lib/src/logic/operations/IPCOperationRunnerPlugin.ts

@@ -81,7 +81,6 @@ export class IPCOperationRunnerPlugin implements IPhasedCommandPlugin {
             initialCommand,
             incrementalCommand,
             commandForHash,
-            persist: true,
             ignoredParameterValues
           });
 

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

libraries/rush-lib/src/logic/operations/OperationGraph.ts

@@ -90,6 +90,23 @@ interface IExecutionIterationContext extends IOperationExecutionRecordContext {
   totalOperations: number;
 }
 
+interface IOperationRunnerCloseEntry {
+  operation: Operation;
+  closeAsync: () => Promise<void>;
+}
+
+class OperationRunnerCloseError extends Error {
+  public readonly operation: Operation;
+  public override readonly cause: Error;
+
+  public constructor(operation: Operation, cause: Error) {
+    super(cause.message, { cause });
+    this.name = OperationRunnerCloseError.name;
+    this.operation = operation;
+    this.cause = cause;
+  }
+}
+
 /**
  * Telemetry data for a phased execution
  */
@@ -390,31 +407,41 @@ export class OperationGraph implements IOperationGraph {
     }
   }
 
-  public async closeRunnersAsync(operations?: Operation[]): Promise<void> {
-    const promises: Promise<void>[] = [];
+  public async closeRunnersAsync(operations?: Iterable<Operation>): Promise<void> {
+    const runnersToClose: IOperationRunnerCloseEntry[] = [];
     const recordMap: ReadonlyMap<Operation, OperationExecutionRecord> =
       this._currentIteration?.records ?? this.resultByOperation;
     const closedRecords: Set<OperationExecutionRecord> = new Set();
     for (const operation of operations ?? this.operations) {
-      if (operation.runner?.closeAsync) {
-        const record: OperationExecutionRecord | undefined = recordMap.get(operation);
-        promises.push(
-          operation.runner.closeAsync().then(() => {
-            if (record) {
-              // Collect for batched notification
-              closedRecords.add(record);
-            }
-          })
-        );
+      const closeAsync: (() => Promise<void>) | undefined = operation.runner?.closeAsync;
+      if (closeAsync) {
+        runnersToClose.push({ operation, closeAsync: closeAsync.bind(operation.runner) });
       }
     }
-    await Promise.all(promises);
-    if (this.abortController.signal.aborted) {
-      return;
+    const results: PromiseSettledResult<void>[] = await Promise.allSettled(
+      runnersToClose.map((entry) => entry.closeAsync())
+    );
+    const errors: OperationRunnerCloseError[] = [];
+    for (let index: number = 0; index < results.length; index++) {
+      const operation: Operation = runnersToClose[index].operation;
+      const result: PromiseSettledResult<void> = results[index];
+      if (result.status === 'fulfilled') {
+        const record: OperationExecutionRecord | undefined = recordMap.get(operation);
+        if (record) {
+          closedRecords.add(record);
+        }
+      } else {
+        const innerError: Error =
+          result.reason instanceof Error ? result.reason : new Error(String(result.reason));
+        errors.push(new OperationRunnerCloseError(operation, innerError));
+      }
     }
-    if (closedRecords.size) {
+    if (!this.abortController.signal.aborted && closedRecords.size) {
       this.hooks.onExecutionStatesUpdated.call(closedRecords);
     }
+    if (errors.length > 0) {
+      throw new AggregateError(errors, `Failed to close ${errors.length} operation runner(s).`);
+    }
   }
 
   public invalidateOperations(operations?: Iterable<Operation>, reason?: string): void {
@@ -873,9 +900,46 @@ 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) {
+        if (e instanceof AggregateError) {
+          for (const error of e.errors) {
+            if (error instanceof OperationRunnerCloseError) {
+              const record: OperationExecutionRecord | undefined = executionRecords.get(error.operation);
+              if (record) {
+                reportRunnerCleanupFailure(record, error.cause);
+              }
+            }
+          }
+        } else {
+          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;