[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;

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/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;
   }
 
@@ -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<OperationStatus> => {
-        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

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

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

@@ -799,6 +799,10 @@ export class OperationGraph implements IOperationGraph {
       onStartAsync: onOperationStartAsync,
       onResultAsync: onOperationCompleteAsync
     };
+    // Track runners closed by normal operation completion so the end-of-iteration fallback does not
+    // issue a concurrent or duplicate closeAsync() call for the same runner.
+    const recordsWithRunnerCleanup: Set<OperationExecutionRecord> = new Set();
+    const graph: OperationGraph = this;
 
     if (!this.quietMode) {
       const plural: string = totalOperations === 1 ? '' : 's';
@@ -873,9 +877,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 +1058,18 @@ export class OperationGraph implements IOperationGraph {
           record.error = e;
           record.status = OperationStatus.Failure;
         }
+        if (!record.shouldRunnerPersist) {
+          // The runner's executeAsync() and the post-operation hook have both settled before cleanup,
+          // so persistence cleanup cannot race this operation's execution.
+          recordsWithRunnerCleanup.add(record);
+          try {
+            await graph.closeRunnersAsync([record.operation]);
+          } catch (e) {
+            _reportOperationErrorIfAny(record);
+            record.error = e;
+            record.status = OperationStatus.Failure;
+          }
+        }
         _onOperationComplete(record, state);
       }
     }