[rush] Fix failures not blocking

common/changes/@microsoft/rush/main_2023-09-21-22-56.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@microsoft/rush",
+      "comment": "Fix a bug in which an operation failing incorrectly does not block its consumers.",
+      "type": "none"
+    }
+  ],
+  "packageName": "@microsoft/rush"
+}

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

@@ -76,14 +76,17 @@ export class AsyncOperationQueue
     this._completedOperations.add(record);
 
     // Apply status changes to direct dependents
-    for (const item of record.consumers) {
-      // Remove this operation from the dependencies, to unblock the scheduler
-      if (
-        item.dependencies.delete(record) &&
-        item.dependencies.size === 0 &&
-        item.status === OperationStatus.Waiting
-      ) {
-        item.status = OperationStatus.Ready;
+    if (record.status !== OperationStatus.Failure && record.status !== OperationStatus.Blocked) {
+      // Only do so if the operation did not fail or get blocked
+      for (const item of record.consumers) {
+        // Remove this operation from the dependencies, to unblock the scheduler
+        if (
+          item.dependencies.delete(record) &&
+          item.dependencies.size === 0 &&
+          item.status === OperationStatus.Waiting
+        ) {
+          item.status = OperationStatus.Ready;
+        }
       }
     }
 

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

@@ -4,7 +4,7 @@
 import colors from 'colors/safe';
 import { TerminalWritable, StdioWritable, TextRewriterTransform } from '@rushstack/terminal';
 import { StreamCollator, CollatedTerminal, CollatedWriter } from '@rushstack/stream-collator';
-import { NewlineKind, Async } from '@rushstack/node-core-library';
+import { NewlineKind, Async, InternalError } from '@rushstack/node-core-library';
 
 import {
   AsyncOperationQueue,
@@ -316,23 +316,28 @@ export class OperationExecutionManager {
         }
         terminal.writeStderrLine(colors.red(`"${name}" failed to build.`));
         const blockedQueue: Set<OperationExecutionRecord> = new Set(record.consumers);
-        for (const blockedRecord of blockedQueue) {
-          if (blockedRecord.status === OperationStatus.Ready) {
-            this._executionQueue.complete(blockedRecord);
-            this._completedOperations++;
 
+        for (const blockedRecord of blockedQueue) {
+          if (blockedRecord.status === OperationStatus.Waiting) {
             // Now that we have the concept of architectural no-ops, we could implement this by replacing
             // {blockedRecord.runner} with a no-op that sets status to Blocked and logs the blocking
             // operations. However, the existing behavior is a bit simpler, so keeping that for now.
             if (!blockedRecord.runner.silent) {
               terminal.writeStdoutLine(`"${blockedRecord.name}" is blocked by "${name}".`);
             }
             blockedRecord.status = OperationStatus.Blocked;
-            this._onOperationStatusChanged?.(blockedRecord);
+
+            this._executionQueue.complete(blockedRecord);
+            this._completedOperations++;
 
             for (const dependent of blockedRecord.consumers) {
               blockedQueue.add(dependent);
             }
+          } else if (blockedRecord.status !== OperationStatus.Blocked) {
+            // It shouldn't be possible for operations to be in any state other than Waiting or Blocked
+            throw new InternalError(
+              `Blocked operation ${blockedRecord.name} is in an unexpected state: ${blockedRecord.status}`
+            );
           }
         }
         this._hasAnyFailures = true;

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

@@ -124,6 +124,42 @@ describe(OperationExecutionManager.name, () => {
     });
   });
 
+  describe('Blocking', () => {
+    it('Failed operations block', async () => {
+      const failingOperation = new Operation({
+        runner: new MockOperationRunner('fail', async () => {
+          return OperationStatus.Failure;
+        })
+      });
+
+      const blockedRunFn: jest.Mock = jest.fn();
+
+      const blockedOperation = new Operation({
+        runner: new MockOperationRunner('blocked', blockedRunFn)
+      });
+
+      blockedOperation.addDependency(failingOperation);
+
+      const manager: OperationExecutionManager = new OperationExecutionManager(
+        new Set([failingOperation, blockedOperation]),
+        {
+          quietMode: false,
+          debugMode: false,
+          parallelism: 1,
+          changedProjectsOnly: false,
+          destination: mockWritable
+        }
+      );
+
+      const result = await manager.executeAsync();
+      expect(result.status).toEqual(OperationStatus.Failure);
+      expect(blockedRunFn).not.toHaveBeenCalled();
+      expect(result.operationResults.size).toEqual(2);
+      expect(result.operationResults.get(failingOperation)?.status).toEqual(OperationStatus.Failure);
+      expect(result.operationResults.get(blockedOperation)?.status).toEqual(OperationStatus.Blocked);
+    });
+  });
+
   describe('Warning logging', () => {
     describe('Fail on warning', () => {
       beforeEach(() => {