[rush] Support aborting phased command execution (#5362)

* [rush] Support aborting phased command execution

* Disambiguate

* Clarify abort special casing.

* Adjust coloring of "aborted" to match "blocked"

---------

Co-authored-by: David Michon <dmichon-msft@users.noreply.github.com>

Author: dmichon-msft

common/changes/@microsoft/rush/rush-abort_2025-09-16-01-25.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@microsoft/rush",
+      "comment": "Support aborting execution in phased commands. The CLI allows aborting via the \"a\" key in watch mode, and it is available to plugin authors for more advanced scenarios.",
+      "type": "none"
+    }
+  ],
+  "packageName": "@microsoft/rush"
+}

common/reviews/api/rush-lib.api.md

@@ -477,6 +477,7 @@ export interface IEnvironmentConfigurationInitializeOptions {
 
 // @alpha
 export interface IExecuteOperationsContext extends ICreateOperationsContext {
+    readonly abortController: AbortController;
     readonly inputsSnapshot?: IInputsSnapshot;
 }
 
@@ -744,6 +745,8 @@ export type IPhaseBehaviorForMissingScript = 'silent' | 'log' | 'error';
 export interface IPhasedCommand extends IRushCommand {
     // @alpha
     readonly hooks: PhasedCommandHooks;
+    // @alpha
+    readonly sessionAbortController: AbortController;
 }
 
 // @public
@@ -1053,6 +1056,7 @@ export class _OperationStateFile {
 
 // @beta
 export enum OperationStatus {
+    Aborted = "ABORTED",
     Blocked = "BLOCKED",
     Executing = "EXECUTING",
     Failure = "FAILURE",

libraries/rush-lib/src/cli/scriptActions/PhasedScriptAction.ts

@@ -98,7 +98,7 @@ interface IRunPhasesOptions extends IInitialRunPhasesOptions {
   executionManagerOptions: IOperationExecutionManagerOptions;
 }
 
-interface IExecutionOperationsOptions {
+interface IExecuteOperationsOptions {
   executeOperationsContext: IExecuteOperationsContext;
   executionManagerOptions: IOperationExecutionManagerOptions;
   ignoreHooks: boolean;
@@ -131,12 +131,13 @@ interface IPhasedCommandTelemetry {
  * and "rebuild" commands are also modeled as phased commands with a single phase that invokes the npm
  * "build" script for each project.
  */
-export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
+export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> implements IPhasedCommand {
   /**
    * @internal
    */
   public _runsBeforeInstall: boolean | undefined;
   public readonly hooks: PhasedCommandHooks;
+  public readonly sessionAbortController: AbortController;
 
   private readonly _enableParallelism: boolean;
   private readonly _isIncrementalBuildAllowed: boolean;
@@ -150,6 +151,7 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
   private readonly _knownPhases: ReadonlyMap<string, IPhase>;
   private readonly _terminal: ITerminal;
   private _changedProjectsOnly: boolean;
+  private _executionAbortController: AbortController | undefined;
 
   private readonly _changedProjectsOnlyParameter: CommandLineFlagParameter | undefined;
   private readonly _selectionParameters: SelectionParameterSet;
@@ -180,6 +182,16 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
     this._runsBeforeInstall = false;
     this._knownPhases = options.phases;
     this._changedProjectsOnly = false;
+    this.sessionAbortController = new AbortController();
+    this._executionAbortController = undefined;
+
+    this.sessionAbortController.signal.addEventListener(
+      'abort',
+      () => {
+        this._executionAbortController?.abort();
+      },
+      { once: true }
+    );
 
     this.hooks = new PhasedCommandHooks();
 
@@ -655,9 +667,11 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
       }
     );
 
+    const abortController: AbortController = (this._executionAbortController = new AbortController());
     const initialExecuteOperationsContext: IExecuteOperationsContext = {
       ...initialCreateOperationsContext,
-      inputsSnapshot: initialSnapshot
+      inputsSnapshot: initialSnapshot,
+      abortController
     };
 
     const executionManagerOptions: IOperationExecutionManagerOptions = {
@@ -670,7 +684,7 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
       }
     };
 
-    const initialOptions: IExecutionOperationsOptions = {
+    const initialOptions: IExecuteOperationsOptions = {
       executeOperationsContext: initialExecuteOperationsContext,
       ignoreHooks: false,
       operations,
@@ -691,14 +705,12 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
     };
   }
 
-  private _registerWatchModeInterface(
-    projectWatcher: ProjectWatcher,
-    abortController: AbortController
-  ): void {
+  private _registerWatchModeInterface(projectWatcher: ProjectWatcher): void {
     const buildOnceKey: 'b' = 'b';
     const changedProjectsOnlyKey: 'c' = 'c';
     const invalidateKey: 'i' = 'i';
     const quitKey: 'q' = 'q';
+    const abortKey: 'a' = 'a';
     const toggleWatcherKey: 'w' = 'w';
     const shutdownProcessesKey: 'x' = 'x';
 
@@ -707,6 +719,7 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
     projectWatcher.setPromptGenerator((isPaused: boolean) => {
       const promptLines: string[] = [
         `  Press <${quitKey}> to gracefully exit.`,
+        `  Press <${abortKey}> to abort queued operations. Any that have started will finish.`,
         `  Press <${toggleWatcherKey}> to ${isPaused ? 'resume' : 'pause'}.`,
         `  Press <${invalidateKey}> to invalidate all projects.`,
         `  Press <${changedProjectsOnlyKey}> to ${
@@ -725,11 +738,15 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
     const onKeyPress = (key: string): void => {
       switch (key) {
         case quitKey:
-          terminal.writeLine(`Exiting watch mode...`);
+          terminal.writeLine(`Exiting watch mode and aborting any scheduled work...`);
           process.stdin.setRawMode(false);
           process.stdin.off('data', onKeyPress);
           process.stdin.unref();
-          abortController.abort();
+          this.sessionAbortController.abort();
+          break;
+        case abortKey:
+          terminal.writeLine(`Aborting current iteration...`);
+          this._executionAbortController?.abort();
           break;
         case toggleWatcherKey:
           if (projectWatcher.isPaused) {
@@ -768,6 +785,7 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
           process.stdin.setRawMode(false);
           process.stdin.off('data', onKeyPress);
           process.stdin.unref();
+          this.sessionAbortController.abort();
           process.kill(process.pid, 'SIGINT');
           break;
       }
@@ -814,8 +832,8 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
       '../../logic/ProjectWatcher'
     );
 
-    const abortController: AbortController = new AbortController();
-    const abortSignal: AbortSignal = abortController.signal;
+    const sessionAbortController: AbortController = this.sessionAbortController;
+    const abortSignal: AbortSignal = sessionAbortController.signal;
 
     const projectWatcher: typeof ProjectWatcher.prototype = new ProjectWatcher({
       getInputsSnapshotAsync,
@@ -829,7 +847,7 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
 
     // Ensure process.stdin allows interactivity before using TTY-only APIs
     if (process.stdin.isTTY) {
-      this._registerWatchModeInterface(projectWatcher, abortController);
+      this._registerWatchModeInterface(projectWatcher);
     }
 
     const onWaitingForChanges = (): void => {
@@ -877,9 +895,13 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
         terminal.writeLine(`    ${Colorize.cyan(name)}`);
       }
 
+      const initialAbortController: AbortController = (this._executionAbortController =
+        new AbortController());
+
       // Account for consumer relationships
       const executeOperationsContext: IExecuteOperationsContext = {
         ...initialCreateOperationsContext,
+        abortController: initialAbortController,
         changedProjectsOnly: !!this._changedProjectsOnly,
         isInitial: false,
         inputsSnapshot: state,
@@ -893,7 +915,7 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
         this.hooks.createOperations.promise(new Set(), executeOperationsContext)
       );
 
-      const executeOptions: IExecutionOperationsOptions = {
+      const executeOptions: IExecuteOperationsOptions = {
         executeOperationsContext,
         // For now, don't run pre-build or post-build in watch mode
         ignoreHooks: true,
@@ -928,29 +950,36 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
   /**
    * Runs a set of operations and reports the results.
    */
-  private async _executeOperationsAsync(options: IExecutionOperationsOptions): Promise<void> {
-    const { executionManagerOptions, ignoreHooks, operations, stopwatch, terminal } = options;
+  private async _executeOperationsAsync(options: IExecuteOperationsOptions): Promise<void> {
+    const {
+      executeOperationsContext,
+      executionManagerOptions,
+      ignoreHooks,
+      operations,
+      stopwatch,
+      terminal
+    } = options;
 
     const executionManager: OperationExecutionManager = new OperationExecutionManager(
       operations,
       executionManagerOptions
     );
 
-    const { isInitial, isWatch } = options.executeOperationsContext;
+    const { isInitial, isWatch, abortController, invalidateOperation } = executeOperationsContext;
 
     let success: boolean = false;
     let result: IExecutionResult | undefined;
 
     try {
       const definiteResult: IExecutionResult = await measureAsyncFn(
         `${PERF_PREFIX}:executeOperationsInner`,
-        () => executionManager.executeAsync()
+        () => executionManager.executeAsync(abortController)
       );
       success = definiteResult.status === OperationStatus.Success;
       result = definiteResult;
 
       await measureAsyncFn(`${PERF_PREFIX}:afterExecuteOperations`, () =>
-        this.hooks.afterExecuteOperations.promise(definiteResult, options.executeOperationsContext)
+        this.hooks.afterExecuteOperations.promise(definiteResult, executeOperationsContext)
       );
 
       stopwatch.stop();
@@ -980,6 +1009,20 @@ export class PhasedScriptAction extends BaseScriptAction<IPhasedCommandConfig> {
       }
     }
 
+    this._executionAbortController = undefined;
+
+    if (invalidateOperation) {
+      const operationResults: ReadonlyMap<Operation, IOperationExecutionResult> | undefined =
+        result?.operationResults;
+      if (operationResults) {
+        for (const [operation, { status }] of operationResults) {
+          if (status === OperationStatus.Aborted) {
+            invalidateOperation(operation, 'aborted');
+          }
+        }
+      }
+    }
+
     if (!ignoreHooks) {
       measureFn(`${PERF_PREFIX}:doAfterTask`, () => this._doAfterTask());
     }

libraries/rush-lib/src/logic/ProjectWatcher.ts

@@ -156,6 +156,13 @@ export class ProjectWatcher {
    * `waitForChange` is not allowed to be called multiple times concurrently.
    */
   public async waitForChangeAsync(onWatchingFiles?: () => void): Promise<IProjectChangeResult> {
+    if (this.isPaused) {
+      this._setStatus(`Project watcher paused.`);
+      await new Promise<void>((resolve) => {
+        this._resolveIfChanged = async () => resolve();
+      });
+    }
+
     const initialChangeResult: IProjectChangeResult = await this._computeChangedAsync();
     // Ensure that the new state is recorded so that we don't loop infinitely
     this._commitChanges(initialChangeResult.inputsSnapshot);

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

@@ -104,7 +104,8 @@ export class AsyncOperationQueue
         record.status === OperationStatus.SuccessWithWarning ||
         record.status === OperationStatus.FromCache ||
         record.status === OperationStatus.NoOp ||
-        record.status === OperationStatus.Failure
+        record.status === OperationStatus.Failure ||
+        record.status === OperationStatus.Aborted
       ) {
         // It shouldn't be on the queue, remove it
         queue.splice(i, 1);

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

@@ -86,7 +86,8 @@ const TIMELINE_CHART_SYMBOLS: Record<OperationStatus, string> = {
   [OperationStatus.Blocked]: '.',
   [OperationStatus.Skipped]: '%',
   [OperationStatus.FromCache]: '%',
-  [OperationStatus.NoOp]: '%'
+  [OperationStatus.NoOp]: '%',
+  [OperationStatus.Aborted]: '@'
 };
 
 const COBUILD_REPORTABLE_STATUSES: Set<OperationStatus> = new Set([
@@ -110,7 +111,8 @@ const TIMELINE_CHART_COLORIZER: Record<OperationStatus, (string: string) => stri
   [OperationStatus.Blocked]: Colorize.red,
   [OperationStatus.Skipped]: Colorize.green,
   [OperationStatus.FromCache]: Colorize.green,
-  [OperationStatus.NoOp]: Colorize.gray
+  [OperationStatus.NoOp]: Colorize.gray,
+  [OperationStatus.Aborted]: Colorize.red
 };
 
 interface ITimelineRecord {

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

@@ -90,6 +90,7 @@ export class OperationExecutionManager {
   // Variables for current status
   private _hasAnyFailures: boolean;
   private _hasAnyNonAllowedWarnings: boolean;
+  private _hasAnyAborted: boolean;
   private _completedOperations: number;
   private _executionQueue: AsyncOperationQueue;
 
@@ -109,6 +110,7 @@ export class OperationExecutionManager {
     this._quietMode = quietMode;
     this._hasAnyFailures = false;
     this._hasAnyNonAllowedWarnings = false;
+    this._hasAnyAborted = false;
     this._parallelism = parallelism;
 
     this._beforeExecuteOperation = beforeExecuteOperation;
@@ -232,9 +234,10 @@ export class OperationExecutionManager {
    * Executes all operations which have been registered, returning a promise which is resolved when all the
    * operations are completed successfully, or rejects when any operation fails.
    */
-  public async executeAsync(): Promise<IExecutionResult> {
+  public async executeAsync(abortController: AbortController): Promise<IExecutionResult> {
     this._completedOperations = 0;
     const totalOperations: number = this._totalOperations;
+    const abortSignal: AbortSignal = abortController.signal;
 
     if (!this._quietMode) {
       const plural: string = totalOperations === 1 ? '' : 's';
@@ -287,10 +290,18 @@ export class OperationExecutionManager {
     await Async.forEachAsync(
       this._executionQueue,
       async (record: OperationExecutionRecord) => {
-        await record.executeAsync({
-          onStart: onOperationStartAsync,
-          onResult: onOperationCompleteAsync
-        });
+        if (abortSignal.aborted) {
+          record.status = OperationStatus.Aborted;
+          // Bypass the normal completion handler, directly mark the operation as aborted and unblock the queue.
+          // We do this to ensure that we aren't messing with the stopwatch or terminal.
+          this._hasAnyAborted = true;
+          this._executionQueue.complete(record);
+        } else {
+          await record.executeAsync({
+            onStart: onOperationStartAsync,
+            onResult: onOperationCompleteAsync
+          });
+        }
       },
       {
         concurrency: maxParallelism,
@@ -300,9 +311,11 @@ export class OperationExecutionManager {
 
     const status: OperationStatus = this._hasAnyFailures
       ? OperationStatus.Failure
-      : this._hasAnyNonAllowedWarnings
-        ? OperationStatus.SuccessWithWarning
-        : OperationStatus.Success;
+      : this._hasAnyAborted
+        ? OperationStatus.Aborted
+        : this._hasAnyNonAllowedWarnings
+          ? OperationStatus.SuccessWithWarning
+          : OperationStatus.Success;
 
     return {
       operationResults: this._executionRecords,
@@ -437,6 +450,11 @@ export class OperationExecutionManager {
         break;
       }
 
+      case OperationStatus.Aborted: {
+        this._hasAnyAborted ||= true;
+        break;
+      }
+
       default: {
         throw new InternalError(`Unexpected operation status: ${status}`);
       }