cobuilds: yield priority to operations that haven't been tried yet

Signed-off-by: Aramis Sennyey <aramissennyeydd@users.noreply.github.com>

Author: aramissennyeydd

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

@@ -22,6 +22,11 @@ export class AsyncOperationQueue
   private readonly _totalOperations: number;
   private readonly _completedOperations: Set<OperationExecutionRecord>;
 
+  // Tracks how many times each operation has been assigned to an execution slot.
+  // Operations that have been assigned more times (e.g. cobuild retries) are sorted
+  // after operations with fewer attempts, so untried work is preferred.
+  private readonly _timesQueued: Map<OperationExecutionRecord, number>;
+
   private _isDone: boolean;
 
   /**
@@ -37,6 +42,7 @@ export class AsyncOperationQueue
     this._totalOperations = this._queue.length;
     this._isDone = false;
     this._completedOperations = new Set<OperationExecutionRecord>();
+    this._timesQueued = new Map();
   }
 
   /**
@@ -63,6 +69,7 @@ export class AsyncOperationQueue
    */
   public complete(record: OperationExecutionRecord): void {
     this._completedOperations.add(record);
+    this._timesQueued.delete(record);
 
     // Apply status changes to direct dependents
     if (record.status !== OperationStatus.Failure && record.status !== OperationStatus.Blocked) {
@@ -86,25 +93,14 @@ export class AsyncOperationQueue
     }
   }
 
-  /**
-   * Moves an operation to the back of the queue (lowest priority).
-   * Used when a cobuild lock acquisition fails, so that locally-executable operations are
-   * assigned before this operation is retried.
-   */
-  public yieldPriority(record: OperationExecutionRecord): void {
-    const index: number = this._queue.indexOf(record);
-    if (index > 0) {
-      this._queue.splice(index, 1);
-      this._queue.unshift(record);
-    }
-  }
-
   /**
    * Routes ready operations with 0 dependencies to waiting iterators. Normally invoked as part of `next()`, but
    * if the caller does not update operation dependencies prior to calling `next()`, may need to be invoked manually.
    */
   public assignOperations(): void {
-    const { _queue: queue, _pendingIterators: waitingIterators } = this;
+    const { _queue: queue, _pendingIterators: waitingIterators, _timesQueued: timesQueued } = this;
+
+    const readyOperations: OperationExecutionRecord[] = [];
 
     // By iterating in reverse order we do less array shuffling when removing operations
     for (let i: number = queue.length - 1; waitingIterators.length > 0 && i >= 0; i--) {
@@ -122,6 +118,7 @@ export class AsyncOperationQueue
       ) {
         // It shouldn't be on the queue, remove it
         queue.splice(i, 1);
+        timesQueued.delete(record);
       } else if (record.status === OperationStatus.Queued || record.status === OperationStatus.Executing) {
         // This operation is currently executing
         // next one plz :)
@@ -133,17 +130,32 @@ export class AsyncOperationQueue
         // Sanity check
         throw new Error(`Unexpected status "${record.status}" for queued operation: ${record.name}`);
       } else {
-        // This task is ready to process, hand it to the iterator.
-        // Needs to have queue semantics, otherwise tools that iterate it get confused
-        record.status = OperationStatus.Queued;
-        waitingIterators.shift()!({
-          value: record,
-          done: false
-        });
+        readyOperations.push(record);
       }
       // Otherwise operation is still waiting
     }
 
+    if (readyOperations.length > 1) {
+      // Sort by times queued ascending. Operations that have never been queued (0)
+      // come first, then operations with fewer attempts. This ensures cobuild retries
+      // (queued 1+ times, returned to Ready) are tried after untried operations.
+      readyOperations.sort((a, b) => (timesQueued.get(a) ?? 0) - (timesQueued.get(b) ?? 0));
+    }
+
+    for (const record of readyOperations) {
+      if (waitingIterators.length === 0) {
+        break;
+      }
+      // This task is ready to process, hand it to the iterator.
+      // Needs to have queue semantics, otherwise tools that iterate it get confused
+      timesQueued.set(record, (timesQueued.get(record) ?? 0) + 1);
+      record.status = OperationStatus.Queued;
+      waitingIterators.shift()!({
+        value: record,
+        done: false
+      });
+    }
+
     // Since items only get removed from the queue when they have a final status, this should be safe.
     if (queue.length === 0) {
       this._isDone = true;

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

@@ -411,7 +411,6 @@ export class CacheableOperationPlugin implements IPhasedCommandPlugin {
               periodicCallback.start();
             } else {
               setTimeout(() => {
-                record.yieldPriority();
                 record.status = OperationStatus.Ready;
               }, 500);
               return OperationStatus.Executing;

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

@@ -148,9 +148,6 @@ export class OperationExecutionManager {
     const executionRecordContext: IOperationExecutionRecordContext = {
       streamCollator: this._streamCollator,
       onOperationStatusChanged: this._onOperationStatusChanged,
-      yieldPriority: (record: OperationExecutionRecord) => {
-        this._executionQueue.yieldPriority(record);
-      },
       createEnvironment: this._createEnvironmentForOperation,
       inputsSnapshot,
       debugMode,

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

@@ -42,7 +42,6 @@ import {
 export interface IOperationExecutionRecordContext {
   streamCollator: StreamCollator;
   onOperationStatusChanged?: (record: OperationExecutionRecord) => void;
-  yieldPriority?: (record: OperationExecutionRecord) => void;
   createEnvironment?: (record: OperationExecutionRecord) => IEnvironment;
   inputsSnapshot: IInputsSnapshot | undefined;
 
@@ -235,14 +234,6 @@ export class OperationExecutionRecord implements IOperationRunnerContext, IOpera
     return !this.operation.enabled || this.runner.silent;
   }
 
-  /**
-   * Moves this operation to the back of the execution queue so that other operations
-   * are assigned first.
-   */
-  public yieldPriority(): void {
-    this._context.yieldPriority?.(this);
-  }
-
   public getStateHash(): string {
     if (this._stateHash === undefined) {
       const components: readonly string[] = this.getStateHashComponents();

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

@@ -169,10 +169,11 @@ describe(AsyncOperationQueue.name, () => {
     expect(result.done).toEqual(true);
   });
 
-  it('yieldPriority moves an operation to the back of the queue', async () => {
+  it('sorts cobuild retries after untried operations', async () => {
     // Three independent operations: A, B, C (all Ready).
-    // A is assigned first. Calling yieldPriority + setting Ready should move it to the back,
-    // so B and C are assigned before A on the next pass.
+    // A is assigned first, then returns to Ready (cobuild lock failed).
+    // On the next pass, B and C should be assigned before A because A
+    // has a recent lastAssignedAt timestamp.
     const opA = createRecord('a');
     const opB = createRecord('b');
     const opC = createRecord('c');
@@ -183,11 +184,10 @@ describe(AsyncOperationQueue.name, () => {
     const r1: IteratorResult<OperationExecutionRecord> = await queue.next();
     const firstAssigned: OperationExecutionRecord = r1.value;
 
-    // Simulate cobuild retry: yield priority then re-ready the operation
-    queue.yieldPriority(firstAssigned);
+    // Simulate cobuild retry: operation returns to Ready
     firstAssigned.status = OperationStatus.Ready;
 
-    // Now assign the remaining — untried operations should come before the retry
+    // Assign all three — untried operations should come before the retry
     const results: OperationExecutionRecord[] = [];
     const r2: IteratorResult<OperationExecutionRecord> = await queue.next();
     results.push(r2.value);
@@ -196,13 +196,13 @@ describe(AsyncOperationQueue.name, () => {
     const r4: IteratorResult<OperationExecutionRecord> = await queue.next();
     results.push(r4.value);
 
-    // The yielded operation should be last
+    // The cobuild retry should be last
     expect(results[2]).toBe(firstAssigned);
   });
 
-  it('yieldPriority assigns freshly unblocked operations first', async () => {
+  it('assigns freshly unblocked operations before cobuild retries', async () => {
     // A (no deps), B (depends on C), C (no deps)
-    // A is assigned and yields priority (cobuild retry).
+    // A is assigned and returns to Ready (cobuild retry).
     // C completes, unblocking B. B should be assigned before A.
     const opA = createRecord('a');
     const opB = createRecord('b');
@@ -217,15 +217,14 @@ describe(AsyncOperationQueue.name, () => {
     const r2: IteratorResult<OperationExecutionRecord> = await queue.next();
     expect(new Set([r1.value, r2.value])).toEqual(new Set([opA, opC]));
 
-    // Simulate: A fails cobuild lock
-    queue.yieldPriority(opA);
+    // Simulate: A fails cobuild lock and returns to Ready
     opA.status = OperationStatus.Ready;
 
     // C succeeds, which unblocks B
     opC.status = OperationStatus.Success;
     queue.complete(opC);
 
-    // B is freshly unblocked, A yielded priority — B should be assigned first
+    // B is freshly unblocked (never assigned), A is a cobuild retry — B should be first
     const r3: IteratorResult<OperationExecutionRecord> = await queue.next();
     expect(r3.value).toBe(opB);